KAFKA-8483/KAFKA-8484; Ensure safe handling of producerId resets by hachikuji · Pull Request #6883 · apache/kafka

hachikuji · 2019-06-05T01:46:44Z

The idempotent producer attempts to detect spurious UNKNOWN_PRODUCER_ID errors and handle them by reassigning sequence numbers to the inflight batches. The inflight batches are tracked in a PriorityQueue. The problem is that the reassignment of sequence numbers depends on the iteration order of PriorityQueue, which does not guarantee any ordering. So this can result in sequence numbers being assigned in the wrong order. This patch fixes the problem by using a sorted set instead of a priority queue so that the iteration order preserves the sequence order. Note that resetting sequence numbers is an exceptional case.

This patch also fixes KAFKA-8484, which can cause an IllegalStateException when the producerId is reset while there are pending produce requests inflight. The solution is to ensure that sequence numbers are only reset if the producerId of a failed batch corresponds to the current producerId.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

hachikuji · 2019-06-05T03:39:00Z

Note this patch also contains a fix to KAFKA-8484. I can separate them if preferable, but I needed some of the common testing logic.

ijuma · 2019-06-08T17:02:35Z

It's fine to include both fixes, can we update the PR title and description to mention that?

ijuma · 2019-06-12T13:26:13Z

I think it would be great to include this in 2.3, if possible.

guozhangwang

LGTM overall. Some minor comments.

guozhangwang · 2019-06-12T15:31:42Z

        }
    }

+    public synchronized void handleCompletedBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response) {


Just to clarify this function is just moved here, no logical changes?

Also if it is indeed the case, could you make some comments when creating the PR for ease of review :) ?

Yes, it is moved to simplify the TransactionManager API so that I could write better test cases. Otherwise it was very difficult to hit the cases without effectively rewriting this logic in the test case.

guozhangwang · 2019-06-12T15:34:30Z

-    synchronized void adjustSequencesDueToFailedBatch(ProducerBatch batch) {
-        if (!topicPartitionBookkeeper.contains(batch.topicPartition))
+    private void adjustSequencesDueToFailedBatch(ProducerBatch batch) {
+        if (!topicPartitionBookkeeper.contains(batch.topicPartition) || !hasProducerIdAndEpoch(batch.producerId(), batch.producerEpoch()))


Why adding the second condition?

This was the fix for KAFKA-8484. However, it seems redundant after I moved things around. Now handleFailedBatch verifies the producerId and epoch, so I think we can remove it.

guozhangwang · 2019-06-12T15:35:04Z

+    }
+
+    public synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
+        maybeTransitionToErrorState(exception);


Same here, the caller function moved here seems not changing any logic (the key change is in adjustSequencesDueToFailedBatch) right?

The main fix is the producerId and epoch check below.

guozhangwang · 2019-06-12T15:38:34Z

-        int sequence = 0;
-        for (ProducerBatch inFlightBatch : topicPartitionBookkeeper.getPartition(topicPartition).inflightBatchesBySequence) {
+    private void startSequencesAtBeginning(TopicPartition topicPartition) {
+        final AtomicInteger sequence = new AtomicInteger(0);


The caller function canRetry is synchronized, do we need an atomic integer?

It was needed only because of the lambda. I guess this is the ugly side of Java 8.

guozhangwang · 2019-06-12T15:50:53Z

+        TransactionManager transactionManager = new TransactionManager();
+        transactionManager.setProducerIdAndEpoch(producerIdAndEpoch);
+
+        ProducerBatch b1 = writeIdempotentBatchWithValue(transactionManager, tp0, "1");


Reading the iterator() code of PriorityQueue, I think three batches are sufficient to expose the randomness of its iterator(). Are there any reasons that you want to have 5, or it's just your favorite magic number?

Not really; 5 seemed like a sufficient interesting number to catch this bug and any future regressions.

…st cases

guozhangwang · 2019-06-12T17:04:08Z

LGTM, feel free to merge after jenkins green.

The idempotent producer attempts to detect spurious UNKNOWN_PRODUCER_ID errors and handle them by reassigning sequence numbers to the inflight batches. The inflight batches are tracked in a PriorityQueue. The problem is that the reassignment of sequence numbers depends on the iteration order of PriorityQueue, which does not guarantee any ordering. So this can result in sequence numbers being assigned in the wrong order. This patch fixes the problem by using a sorted set instead of a priority queue so that the iteration order preserves the sequence order. Note that resetting sequence numbers is an exceptional case. This patch also fixes KAFKA-8484, which can cause an IllegalStateException when the producerId is reset while there are pending produce requests inflight. The solution is to ensure that sequence numbers are only reset if the producerId of a failed batch corresponds to the current producerId. Reviewers: Guozhang Wang <wangguoz@gmail.com>

…che#6883) The idempotent producer attempts to detect spurious UNKNOWN_PRODUCER_ID errors and handle them by reassigning sequence numbers to the inflight batches. The inflight batches are tracked in a PriorityQueue. The problem is that the reassignment of sequence numbers depends on the iteration order of PriorityQueue, which does not guarantee any ordering. So this can result in sequence numbers being assigned in the wrong order. This patch fixes the problem by using a sorted set instead of a priority queue so that the iteration order preserves the sequence order. Note that resetting sequence numbers is an exceptional case. This patch also fixes KAFKA-8484, which can cause an IllegalStateException when the producerId is reset while there are pending produce requests inflight. The solution is to ensure that sequence numbers are only reset if the producerId of a failed batch corresponds to the current producerId. Reviewers: Guozhang Wang <wangguoz@gmail.com>

hachikuji force-pushed the KAFKA-8483 branch 2 times, most recently from 4bd23f9 to d7d712c Compare June 5, 2019 17:58

ijuma requested a review from guozhangwang June 8, 2019 17:03

hachikuji changed the title ~~KAFKA-8483; Ensure message ordering is preserved after sequence resets~~ KAFKA-8483/KAFKA-8484; Ensure safe handling of producerId resets Jun 11, 2019

guozhangwang reviewed Jun 12, 2019

View reviewed changes

hachikuji added 6 commits June 12, 2019 09:27

KAFKA-8483; Ensure message ordering is preserved after sequence resets

e0f6910

Move completion/failure logic into TransactionManager and clean up te…

3eecd6b

…st cases

Fix check for responses following producerId reset

349de80

Need to protect batch completion as well

2b2e653

Fix failing test

e3727a8

Remove redundant producerId and epoch check in adjustSequenceNumbers

14e75b6

hachikuji force-pushed the KAFKA-8483 branch from 4a918bc to 14e75b6 Compare June 12, 2019 16:30

hachikuji merged commit af28010 into apache:trunk Jun 12, 2019

ddrid mentioned this pull request Apr 5, 2022

KAFKA-13794: Fix comparator of inflightBatchesBySequence in TransactionManager #11991

Merged

3 tasks

Conversation

hachikuji commented Jun 5, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji commented Jun 5, 2019

Uh oh!

ijuma commented Jun 8, 2019

Uh oh!

ijuma commented Jun 12, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jun 12, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

hachikuji commented Jun 5, 2019 •

edited

Loading