[GitHub] [kafka] junrao commented on a diff in pull request #14111: KAFKA-9800: Exponential backoff for Kafka clients - KIP-580

["junrao (via GitHub)" <gi...@apache.org>]

junrao commented on code in PR #14111:
URL: https://github.com/apache/kafka/pull/14111#discussion_r1282312134


##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -73,6 +75,7 @@ public class RecordAccumulator {
     private final CompressionType compression;
     private final int lingerMs;
     private final long retryBackoffMs;
+    private final ExponentialBackoff retryBackoff;

Review Comment:
   There is still one reference of retryBackoffMs in batchReady(). Should that be changed with exponential backoff?



##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java:
##########
@@ -384,6 +389,32 @@ int attempts() {
         return attempts.get();
     }
 
+    /*
+     * Returns whether the leader Node has changed since the last attempt.
+     * @param node The Node currently thought of as the leader, which might be null.
+     * @return true if the leader has changed, otherwise false
+     */
+    boolean hasLeaderChanged(Node latestLeader) {
+        boolean leaderChanged = false;
+        if (latestLeader != null) {
+            // If we don't know the leader yet, we have not yet attempted to send to the leader
+            if (currentLeader == null) {
+                currentLeader = latestLeader;
+            } else {
+                // If the leader's node id has changed, this counts as a leader change
+                if (currentLeader.id() != latestLeader.id()) {

Review Comment:
   We probably need to take leaderEpoch into consideration. For example, if the leader epoch has changed, even if the leader node remains the same, it's still worth to retry the batch now.



##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -701,7 +713,9 @@ private long partitionReady(Cluster cluster, long nowMs, String topic,
                 }
 
                 waitedTimeMs = batch.waitedTimeMs(nowMs);
-                backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;
+                backingOff = !batch.hasLeaderChanged(leader) &&
+                        batch.attempts() > 0 &&
+                        waitedTimeMs < retryBackoff.backoff(batch.attempts() - 1);

Review Comment:
   This mostly duplicates the code in line 716. Could we pull that into a util method and reuse?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org