[GitHub] [kafka] hachikuji commented on a change in pull request #9832: KAFKA-12152; Idempotent Producer does not reset the sequence number of partitions without in-flight batches

[GitBox <gi...@apache.org>]

hachikuji commented on a change in pull request #9832:
URL: https://github.com/apache/kafka/pull/9832#discussion_r556208798



##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
##########
@@ -567,8 +571,18 @@ private void bumpIdempotentProducerEpoch() {
             this.topicPartitionBookkeeper.startSequencesAtBeginning(topicPartition, this.producerIdAndEpoch);
             this.partitionsWithUnresolvedSequences.remove(topicPartition);
         }
-
         this.partitionsToRewriteSequences.clear();
+
+        // When the epoch is bumped, reset the sequences for the partitions(s) that do not
+        // have in-flight batches. Sequences of the in-flight batches that did not triggered
+        // the epoch bump are treated when their last in-flight batch is completed.
+        for (TopicPartition topicPartition : this.topicPartitionBookkeeper.partitions()) {

Review comment:
       Apologies for the lengthy comment, but it always takes me some time to wrap my head around the logic here. I'm trying to convince myself whether it is necessary to be selective or if we can reset the sequences for all partitions blindly. Suppose that we are producing to 3 topic partitions and we have the following local state:
   
   ```
   epoch: 5
   ------------
   partition: 0
   last acked seq: 10
   next seq: 11
   inflight seqs: []
   ------------
   partition: 1
   last acked seq: 20
   next seq: 23
   inflight seqs: [21, 22]
   ------------
   partition: 2
   last acked seq: 30
   next seq: 33
   inflight seqs: [31, 32]
   ```
   
   So we have one partition (0) with no inflights and two partitions (1 and 2) which have two batches in-flight each. Suppose that the request with seq 31 for partition 2 returns with an out of sequence error, which causes a call to `requestEpochBumpForPartition`. This in turn causes a call to `bumpIdempotentProducerEpoch`. I think it's clear that we want to reset the sequence number for partition 0 as you are doing here. So we end up with the following state for partition 0:
   ```
   partition: 0
   last acked seq: -1
   next seq: 0
   inflight seqs: []
   ```
   
   For partition 2, the current logic assumes that seq 32 is doomed to fail since 31 has already failed, so we will reset the sequence numbers. In `TransactionManager.canRetry`, we have logic to check that the sequence number has been reset after an out of sequence error and to allow the retry. This leaves us with the following state:
   ```
   partition: 2
   last acked seq: -1
   next seq: 2
   inflight seqs: [0, 1]
   ```
   
   The challenge for partition 1 is that we do not want to alter the sequence numbers of inflight requests unless we have to since that could introduce duplicates if they ultimately need to be retried. As far as I can tell, the following cases are possible:
   
   Case 1: inflight sequence numbers 21 and 22 return successfully
   
   We have the following check in `handleCompletedBatch`:
   ```java
           if (producerIdOrEpochNotMatch(batch) && !hasInflightBatches(batch.topicPartition)) {
               // If the batch was on a different ID and/or epoch (due to an epoch bump) and all its in-flight batches
               // have completed, reset the partition sequence so that the next batch (with the new epoch) starts from 0
               topicPartitionBookkeeper.startSequencesAtBeginning(batch.topicPartition, this.producerIdAndEpoch);
           }
   ```
   So basically we wait until all of the batches have completed, and then reset the next sequence. This leaves us with this state:
   ```
   partition: 1
   last acked seq: -1
   next seq: 0
   inflight seqs: []
   ```
   
   Case 2: one of the inflight batches fails
   
   This case is harder to track down because of the logic in `canRetry`, but suppose that we eventually reach `handleFailedBatch`. I think we will hit the following case:
   ```java
           if (producerIdOrEpochNotMatch(batch)) {
               log.debug("Ignoring failed batch {} with producer id {}, epoch {}, and sequence number {} " +
                               "since the producerId has been reset internally", batch, batch.producerId(),
                       batch.producerEpoch(), batch.baseSequence(), exception);
               return;
           }
   ```
   Because the epoch from the batch does not match, we just discard it. Unless I am missing something, that leaves us in the following state:
   ```
   partition: 1
   last acked seq: 20
   next seq: 23
   inflight seqs: []
   ```
   Then the next produce request to this partition is likely to get an out of sequence error because of a similar problem to the one you found. (Please check this over and see if you reach the same conclusion.)
   
   
   So coming back to the original question, I am trying to think how to simplify the logic here and fix some of the unhandled cases. It's very difficult to reason about the state when we allow inflight requests from different epochs, but I don't see a way around it. Here is an idea how we can improve this logic:
   
   1. I think we need to track the producer epoch and producer id inside each `TopicPartitionEntry`. Our lives will be a lot easier if we can ensure that `nextSequence`, `lastAckedSequence`, and `inflightBatchesBySequence` are all consistently aligned on the same producerId/epoch. 
   2. As the logic does now, we want to drain inflight requests before we begin using the bumped producerId/epoch. This is used to enforce the invariant in 1.
   3. If we hit an out of sequence error, we can basically check whether the producerId/epoch in `TopicPartitionEntry` match the latest one in use by the producer. If they do, then we request a bump and reset sequence numbers. If not, then we can use the latest producerId/epoch and reset sequence numbers.
   4. When we attempt to increment the sequence number after draining batches in `RecordAccumulator.drainBatchesForOneNode`, we can check whether the producerId/epoch in `TopicPartitionEntry` matches the latest value. If it does, then we proceed normally. If it does not, then we check inflight requests. If there are still some remaining, then we cannot drain. If there are no inflights, then we can update producer state.
   
   This probably needs a bit of refinement, but hopefully the core idea is clear. Basically we add producerId/epoch to `TopicPartitionEntry` and we keep the state consistently aligned. Then we lazily update to the latest value once all the inflights have cleared.




----------------------------------------------------------------
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.

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