Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 2 commits into from
Apr 5, 2022
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
78985b4
fix comparator in TransactionManager
Apr 3, 2022
b4d2e86
fix review comments
Apr 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,16 +184,22 @@ private static class TopicPartitionEntry {
// responses which are due to the retention period elapsing, and those which are due to actual lost data.
private long lastAckedOffset;

private static final Comparator<ProducerBatch> PRODUCER_BATCH_COMPARATOR = (b1, b2) -> {
if (b1.baseSequence() < b2.baseSequence()) return -1;
else if (b1.baseSequence() > b2.baseSequence()) return 1;
else return b1.equals(b2) ? 0 : 1;

@artemlivshits artemlivshits Apr 5, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wouldn't this violate the requirements for the compare method?

The implementor must ensure that sgn(compare(x, y)) == -sgn(compare(y, x)) for all x and y. (This implies that compare(x, y) must throw an exception if and only if compare(y, x) throws an exception.)

The implementor must also ensure that the relation is transitive: ((compare(x, y)>0) && (compare(y, z)>0)) implies compare(x, z)>0.

Objects that are not equal need to have a stable order otherwise, binary search may not find the objects.

@ddrid ddrid Apr 6, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @artemlivshits, thanks for your comment. I don't think it violate the requirements for the compare method since we are comparing two batches using an integer.
As for the stable order, I think it doesn't affect the current code, but I can fix this in another pr if you regard it necessary. What do you think?

@artemlivshits artemlivshits Apr 6, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Say we have 2 batches b1 and b2 that have the same base sequence, but are not equal. Then compare(b1, b2) == 1 and compare(b2, b1) == 1, which violates the requirement of changing the sign when the argument order is changed.
This property is used in binary search tree to order and search elements, if it's violated, then we may not find the element because we follow the wrong branch. Say we have some elements in logical order a, b1, b2, x, y when we ordered them we did compare(b2, b1), which returned 1 meaning that b2 is greater than b1. When we search for b1, we may start with b2 and use compare(b1, b2), which would return 1 meaning that b1 is greater than b2, and we continue searching in the x, y part and conclude that it's not there.

@ddrid ddrid Apr 7, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@artemlivshits Thanks for the explanation! I think it make sense. I will open another pr to fix it

@showuon showuon Apr 7, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch! @artemlivshits !

};

TopicPartitionEntry() {
this.producerIdAndEpoch = ProducerIdAndEpoch.NONE;
this.nextSequence = 0;
this.lastAckedSequence = NO_LAST_ACKED_SEQUENCE_NUMBER;
this.lastAckedOffset = ProduceResponse.INVALID_OFFSET;
this.inflightBatchesBySequence = new TreeSet<>(Comparator.comparingInt(ProducerBatch::baseSequence));
this.inflightBatchesBySequence = new TreeSet<>(PRODUCER_BATCH_COMPARATOR);
}

void resetSequenceNumbers(Consumer<ProducerBatch> resetSequence) {
TreeSet<ProducerBatch> newInflights = new TreeSet<>(Comparator.comparingInt(ProducerBatch::baseSequence));
TreeSet<ProducerBatch> newInflights = new TreeSet<>(PRODUCER_BATCH_COMPARATOR);
for (ProducerBatch inflightBatch : inflightBatchesBySequence) {
resetSequence.accept(inflightBatch);
newInflights.add(inflightBatch);
Expand Down
62 changes: 62 additions & 0 deletions ...nts/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -674,6 +674,68 @@ public void testBatchCompletedAfterProducerReset() {
assertNull(transactionManager.nextBatchBySequence(tp0));
}

@Test
public void testDuplicateSequenceAfterProducerReset() throws Exception {
initializeTransactionManager(Optional.empty());
initializeIdempotentProducerId(producerId, epoch);

Metrics metrics = new Metrics(time);
final int requestTimeout = 10000;
final int deliveryTimeout = 15000;

RecordAccumulator accumulator = new RecordAccumulator(logContext, 16 * 1024, CompressionType.NONE, 0, 0L,
deliveryTimeout, metrics, "", time, apiVersions, transactionManager,
new BufferPool(1024 * 1024, 16 * 1024, metrics, time, ""));

Sender sender = new Sender(logContext, this.client, this.metadata, accumulator, false,
MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(metrics), this.time, requestTimeout,
0, transactionManager, apiVersions);

assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());

Future<RecordMetadata> responseFuture1 = accumulator.append(tp0, time.milliseconds(), "1".getBytes(), "1".getBytes(), Record.EMPTY_HEADERS,
null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
sender.runOnce();
assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());

time.sleep(requestTimeout);
sender.runOnce();
assertEquals(0, client.inFlightRequestCount());
assertTrue(transactionManager.hasInflightBatches(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());
sender.runOnce(); // retry
assertEquals(1, client.inFlightRequestCount());
assertTrue(transactionManager.hasInflightBatches(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());

time.sleep(5000); // delivery time out
sender.runOnce();

// The retried request will remain inflight until the request timeout
// is reached even though the delivery timeout has expired and the
// future has completed exceptionally.
assertTrue(responseFuture1.isDone());
TestUtils.assertFutureThrows(responseFuture1, TimeoutException.class);
assertFalse(transactionManager.hasInFlightRequest());

@hachikuji hachikuji Apr 5, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The behavior here puzzled me a little when I was trying to understand the test. Would a comment like this help?

// The retried request will remain inflight until the request timeout
// is reached even though the delivery timeout has expired and the
// future has completed exceptionally.

assertEquals(1, client.inFlightRequestCount());

sender.runOnce(); // bump the epoch
assertEquals(epoch + 1, transactionManager.producerIdAndEpoch().epoch);
assertEquals(0, transactionManager.sequenceNumber(tp0).intValue());

Future<RecordMetadata> responseFuture2 = accumulator.append(tp0, time.milliseconds(), "2".getBytes(), "2".getBytes(), Record.EMPTY_HEADERS,
null, MAX_BLOCK_TIMEOUT, false, time.milliseconds()).future;
sender.runOnce();
sender.runOnce();
assertEquals(0, transactionManager.firstInFlightSequence(tp0));
assertEquals(1, transactionManager.sequenceNumber(tp0).intValue());

time.sleep(5000); // request time out again
sender.runOnce();
assertTrue(transactionManager.hasInflightBatches(tp0)); // the latter batch failed and retried
assertFalse(responseFuture2.isDone());
}

private ProducerBatch writeIdempotentBatchWithValue(TransactionManager manager,
TopicPartition tp,
String value) {
Expand Down