private boolean shouldBackoff(boolean hasLeaderChanged, final ProducerBatch batch, final long waitedTimeMs) {
        boolean shouldWaitMore = batch.attempts() > 0 && waitedTimeMs < retryBackoff.backoff(batch.attempts() - 1);
        boolean shouldBackoff = !hasLeaderChanged && shouldWaitMore;
        if (log.isTraceEnabled()) {
            if (shouldBackoff) {
                log.trace(
                    "For {}, will backoff", batch);
            } else {
                log.trace(
                    "For {}, will not backoff, shouldWaitMore {}, hasLeaderChanged {}", batch,
                    shouldWaitMore, hasLeaderChanged);
            }
        } else if (log.isDebugEnabled() && hasLeaderChanged) {
            // Add less-verbose log at DEBUG.
            log.debug("For {}, leader has changed, hence skipping backoff.", batch);
        }
        return shouldBackoff;
    }
Check whether there are any batches which haven't been drained
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java