import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
private void doAutoCommitOffsetsAsync() {
    Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
    log.debug("Sending asynchronous auto-commit of offsets {} for group {}", allConsumedOffsets, groupId);
    //rebalance
    commitOffsetsAsync(allConsumedOffsets, new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
            if (exception != null) {
                log.warn("Auto-commit of offsets {} failed for group {}: {}", offsets, groupId,
                        exception.getMessage());
                if (exception instanceof RetriableException)
                    nextAutoCommitDeadline = Math.min(time.milliseconds() + retryBackoffMs, nextAutoCommitDeadline);
            } else {
                log.debug("Completed auto-commit of offsets {} for group {}", offsets, groupId);
            }
        }
    });
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
private void doCommitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, final OffsetCommitCallback callback) {
    this.subscriptions.needRefreshCommits();
    //rebalance
    RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
    final OffsetCommitCallback cb = callback == null ? defaultOffsetCommitCallback : callback;
    future.addListener(new RequestFutureListener<Void>() {
        @Override
        public void onSuccess(Void value) {
            if (interceptors != null)
                interceptors.onCommit(offsets);

            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, null));
        }

        @Override
        public void onFailure(RuntimeException e) {
            Exception commitException = e;

            if (e instanceof RetriableException)
                commitException = RetriableCommitFailedException.withUnderlyingMessage(e.getMessage());

            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
        }
    });
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
private <V extends AbstractStatus> void send(final String key,
                                             final V status,
                                             final CacheEntry<V> entry,
                                             final boolean safeWrite) {
    final int sequence;
    synchronized (this) {
        this.generation = status.generation();
        if (safeWrite && !entry.canWriteSafely(status))
            return;
        sequence = entry.increment();
    }

    final byte[] value = status.state() == ConnectorStatus.State.DESTROYED ? null : serialize(status);

    kafkaLog.send(key, value, new org.apache.kafka.clients.producer.Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
                if (exception instanceof RetriableException) {
                    synchronized (KafkaStatusBackingStore.this) {
                        if (entry.isDeleted()
                                || status.generation() != generation
                                || (safeWrite && !entry.canWriteSafely(status, sequence)))
                            return;
                    }
                    kafkaLog.send(key, value, this);
                } else {
                    log.error("Failed to write status update", exception);
                }
            }
        }
    });
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
public void commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
    this.subscriptions.needRefreshCommits();
    RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
    final OffsetCommitCallback cb = callback == null ? defaultOffsetCommitCallback : callback;
    future.addListener(new RequestFutureListener<Void>() {
        @Override
        public void onSuccess(Void value) {
            if (interceptors != null)
                interceptors.onCommit(offsets);
            cb.onComplete(offsets, null);
        }

        @Override
        public void onFailure(RuntimeException e) {
            if (e instanceof RetriableException) {
                cb.onComplete(offsets, new RetriableCommitFailedException("Commit offsets failed with retriable exception. You should retry committing offsets.", e));
            } else {
                cb.onComplete(offsets, e);
            }
        }
    });

    // ensure the commit has a chance to be transmitted (without blocking on its completion).
    // Note that commits are treated as heartbeats by the coordinator, so there is no need to
    // explicitly allow heartbeats through delayed task execution.
    client.pollNoWakeup();
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
private RetryStrategy strategy() {
    return Retries
        .defaultStrategy(onStop)
        .retryException(RetriableException.class);
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
/**
 * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed
 */
private boolean canRetry(ProducerBatch batch, Errors error) {
    return batch.attempts() < this.retries && error.exception() instanceof RetriableException;
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
/**
 * Handle a failure.
 *
 * Depending on what the exception is and how many times we have already tried, we may choose to
 * fail the Call, or retry it.  It is important to print the stack traces here in some cases,
 * since they are not necessarily preserved in ApiVersionException objects.
 *
 * @param now           The current time in milliseconds.
 * @param throwable     The failure exception.
 */
final void fail(long now, Throwable throwable) {
    // If this is an UnsupportedVersionException that we can retry, do so.
    if ((throwable instanceof UnsupportedVersionException) &&
             handleUnsupportedVersionException((UnsupportedVersionException) throwable)) {
        log.trace("{} attempting protocol downgrade.", this);
        runnable.enqueue(this, now);
        return;
    }
    tries++;
    // If the call has timed out, fail.
    if (calcTimeoutMsRemainingAsInt(now, deadlineMs) < 0) {
        if (log.isDebugEnabled()) {
            log.debug("{} timed out at {} after {} attempt(s)", this, now, tries,
                new Exception(prettyPrintException(throwable)));
        }
        handleFailure(throwable);
        return;
    }
    // If the exception is not retryable, fail.
    if (!(throwable instanceof RetriableException)) {
        if (log.isDebugEnabled()) {
            log.debug("{} failed with non-retriable exception after {} attempt(s)", this, tries,
                new Exception(prettyPrintException(throwable)));
        }
        handleFailure(throwable);
        return;
    }
    // If we are out of retries, fail.
    if (tries > maxRetries) {
        if (log.isDebugEnabled()) {
            log.debug("{} failed after {} attempt(s)", this, tries,
                new Exception(prettyPrintException(throwable)));
        }
        handleFailure(throwable);
        return;
    }
    if (log.isDebugEnabled()) {
        log.debug("{} failed: {}.  Beginning retry #{}",
            this, prettyPrintException(throwable), tries);
    }
    runnable.enqueue(this, now);
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
/**
 * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed
 */
private boolean canRetry(RecordBatch batch, Errors error) {
    return batch.attempts < this.retries && error.exception() instanceof RetriableException;
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
/**
 * Check if the request is retriable (convenience method for checking if
 * the exception is an instance of {@link RetriableException}.
 * @return true if it is retriable, false otherwise
 * @throws IllegalStateException if the future is not complete or completed successfully
 */
public boolean isRetriable() {
    return exception() instanceof RetriableException;
}
 

import org.apache.kafka.common.errors.RetriableException; //导入依赖的package包/类
/**
 * Check if the request is retriable (convenience method for checking if
 * the exception is an instance of {@link RetriableException}.
 * @return true if it is retriable, false otherwise
 */
public boolean isRetriable() {
    return exception instanceof RetriableException;
}