chia7712 commented on code in PR #23663:
URL: https://github.com/apache/kafka/pull/23663#discussion_r4288563702
##########
tests/kafkatest/tests/core/reassign_partitions_test.py:
##########
@@ -163,8 +163,7 @@ def test_reassign_partitions(self, bounce_brokers,
reassign_from_offset_zero, me
self.producer = VerifiableProducer(self.test_context,
self.num_producers,
self.kafka, self.topic,
throughput=self.producer_throughput,
- # Once KAFKA-18905 is fixed, the
idempotent producer should be enabled.
- enable_idempotence=False,
+ enable_idempotence=True,
Review Comment:
Have you run this e2e in a loop to make sure it is not flaky anymore?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -97,6 +103,31 @@ boolean hasInflightBatches() {
return !inflightBatchesBySequence.isEmpty();
}
+ /**
+ * Returns the number of batches sent after the oldest in-flight batch,
completed or not.
+ * <p>
+ * As a side effect this forgets completed batches that are no longer
ahead of any in-flight batch. That is done
+ * here rather than when a batch is removed from the in-flight set,
because a batch that is rejected as too large
+ * is removed and then re-added as several smaller batches with the
sequences it had before, so at removal time
+ * it is not yet known which completed batches are still ahead of
something in flight.
+ */
+ int numBatchesAheadOfOldestInflight() {
+ forgetCompletedSequencesNotAheadOfInflightBatches();
+ if (inflightBatchesBySequence.isEmpty()) {
+ return 0;
+ }
+ return inflightBatchesBySequence.size() - 1 +
completedSequencesAheadOfOldestInflight.size();
+ }
+
+ private void forgetCompletedSequencesNotAheadOfInflightBatches() {
Review Comment:
Could you merge both methods?
```java
int numBatchesAheadOfOldestInflight() {
if (inflightBatchesBySequence.isEmpty()) {
completedSequencesAheadOfOldestInflight.clear();
return 0;
}
int oldestSequence = inflightBatchesBySequence.first().baseSequence();
completedSequencesAheadOfOldestInflight.headSet(oldestSequence,
true).clear();
return inflightBatchesBySequence.size() - 1 +
completedSequencesAheadOfOldestInflight.size();
}
```
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -133,7 +171,17 @@ int maybeUpdateLastAckedSequence(int sequence) {
}
void removeInFlightBatch(ProducerBatch batch) {
- inflightBatchesBySequence.remove(batch);
+ if (!inflightBatchesBySequence.remove(batch)) {
+ return;
+ }
+ if (!inflightBatchesBySequence.isEmpty()
+ && batch.baseSequence() >
inflightBatchesBySequence.first().baseSequence()) {
Review Comment:
Do we really need the condition `batch.baseSequence() >
inflightBatchesBySequence.first().baseSequence()`? We eventually prune the
collection when calling `numBatchesAheadOfOldestInflight`
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -133,7 +167,14 @@ int maybeUpdateLastAckedSequence(int sequence) {
}
void removeInFlightBatch(ProducerBatch batch) {
- inflightBatchesBySequence.remove(batch);
+ if (!inflightBatchesBySequence.remove(batch)) {
+ return;
+ }
+ if (!inflightBatchesBySequence.isEmpty()
+ && batch.baseSequence() >
inflightBatchesBySequence.first().baseSequence()) {
+ // Completed while an older batch is still in flight, see
`completedSequencesAheadOfOldestInflight`.
+ completedSequencesAheadOfOldestInflight.add(batch.baseSequence());
Review Comment:
Since a batch removed here can be added back by `addInflightBatch` with the
same base sequence (that is what happens when a batch is split),
`addInflightBatch` should remove the sequence from
`completedSequencesAheadOfOldestInflight`, right?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -149,6 +190,17 @@ void adjustSequencesDueToFailedBatch(long baseSequence,
int recordCount) {
inFlightBatch.resetProducerState(new
ProducerIdAndEpoch(inFlightBatch.producerId(), inFlightBatch.producerEpoch()),
newSequence);
});
+
+ // The failed batch was never appended, so it no longer occupies a
sequence range, and the completed batches
+ // after it move down along with the in-flight ones. The failed batch
itself may have been recorded as
+ // completed ahead of the oldest in-flight batch when it was removed
from the in-flight set, so drop it first.
+ completedSequencesAheadOfOldestInflight.remove((int) baseSequence);
Review Comment:
the type of `baseSequence` should be `int` rather than `long` ... would you
mind fixing the method signature?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -149,6 +190,17 @@ void adjustSequencesDueToFailedBatch(long baseSequence,
int recordCount) {
inFlightBatch.resetProducerState(new
ProducerIdAndEpoch(inFlightBatch.producerId(), inFlightBatch.producerEpoch()),
newSequence);
});
+
+ // The failed batch was never appended, so it no longer occupies a
sequence range, and the completed batches
+ // after it move down along with the in-flight ones. The failed batch
itself may have been recorded as
+ // completed ahead of the oldest in-flight batch when it was removed
from the in-flight set, so drop it first.
+ completedSequencesAheadOfOldestInflight.remove((int) baseSequence);
+ TreeSet<Integer> adjustedSequences = new TreeSet<>();
Review Comment:
We just collect the shifted seqs here and put them back into the sorted set,
so the temporary `TreeSet` can be a plain List.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
