You are viewing a plain text version of this content. The canonical link for it is here.

Posted to jira@kafka.apache.org by "jolshan (via GitHub)" <gi...@apache.org> on 2023/04/18 20:40:21 UTC

[GitHub] [kafka] jolshan opened a new pull request, #13608: KAFKA-14844: Include check transaction is still ongoing right before append

jolshan opened a new pull request, #13608:
URL: https://github.com/apache/kafka/pull/13608

   We will need to pick up the changes in KAFKA-14916 (right now we assume producer ID is shared in all batches), but wanted to get a WIP draft out for general ideas.
   
   Will need to add the commented section in analyzeAndValidateProducerState and tests
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


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

[GitHub] [kafka] artemlivshits commented on a diff in pull request #13608: KAFKA-14884: Include check transaction is still ongoing right before append

Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.

artemlivshits commented on code in PR #13608:
URL: https://github.com/apache/kafka/pull/13608#discussion_r1197145142


##########
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateEntry.java:
##########
@@ -41,18 +42,35 @@ public class ProducerStateEntry {
     private int coordinatorEpoch;
     private long lastTimestamp;
     private OptionalLong currentTxnFirstOffset;
+    
+    private VerificationState verificationState;
+    
+    // Before any batches are associated with the entry, the tentative sequence represents the lowest sequence seen.
+    private OptionalInt tentativeSequence;
+    
+    public enum VerificationState {
+        EMPTY,
+        VERIFYING,
+        VERIFIED
+    }

Review Comment:
   As discussed offline, this race can cause hanging transactions:
   1. producer properly adds partition to transaction
   2. producer has trouble sending messages, retries a few times (all messages have the same epoch + sequence)
   3. partition leader gets first message, sets state to VERIFYING, issues request to TC
   4. TC sees the open transaction, replies with success
   5. producer gives up and aborts transaction
   6. partition leader gets abort marker, sets state to EMPTY
   7. partition leader gets second message, sets state to VERIFYING, issues request to TC
   8. successful reply from step 4 finally arrives, sees the state is VERIFYING, sets to VERIFIED
   9. the first message is written on top of abort marker, resulting in "hanging" transaction
   
   the issue is that transition VERIFYING - EMPTY - VERIFYING is the same as VERIFYING, the classic ABA problem https://en.wikipedia.org/wiki/ABA_problem



##########
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateEntry.java:
##########
@@ -104,27 +122,63 @@ public boolean maybeUpdateProducerEpoch(short producerEpoch) {
         }
     }
 
+    public boolean maybeUpdateProducerHigherEpoch(short producerEpoch) {
+        if (this.producerEpoch < producerEpoch) {
+            batchMetadata.clear();
+            this.producerEpoch = producerEpoch;
+            return true;
+        } else {
+            return false;
+        }
+    }
+    
+    // We only set tentative sequence if no batches have been written to the log. It is used to avoid OutOfOrderSequenceExceptions
+    // when we saw a lower sequence during transaction verification. We will update the sequence when there is no batch metadata if:
+    //  a) There is no tentative sequence yet
+    //  b) A lower sequence for the same epoch is seen and should thereby block records after that
+    //  c) A higher producer epoch is found that will reset the lowest seen sequence
+    public void maybeUpdateTentativeSequence(int sequence, short producerEpoch) {
+        if (batchMetadata.isEmpty() && 

Review Comment:
   We only need tentative state when we don't have an entry for something that we've written.  I'm not sure if doing something that we don't need so that we don't rely on it later makes the code simpler :-).



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