jolshan commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375118370
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -55,20 +63,111 @@ class TxnPartitionEntry {
.thenComparingInt(ProducerBatch::producerEpoch)
.thenComparingInt(ProducerBatch::baseSequence);
- TxnPartitionEntry() {
+ TxnPartitionEntry(TopicPartition topicPartition) {
+ this.topicPartition = topicPartition;
this.producerIdAndEpoch = ProducerIdAndEpoch.NONE;
this.nextSequence = 0;
- this.lastAckedSequence =
TransactionManager.NO_LAST_ACKED_SEQUENCE_NUMBER;
+ this.lastAckedSequence = NO_LAST_ACKED_SEQUENCE_NUMBER;
this.lastAckedOffset = ProduceResponse.INVALID_OFFSET;
this.inflightBatchesBySequence = new
TreeSet<>(PRODUCER_BATCH_COMPARATOR);
}
- void resetSequenceNumbers(Consumer<ProducerBatch> resetSequence) {
+ ProducerIdAndEpoch producerIdAndEpoch() {
+ return producerIdAndEpoch;
+ }
+
+ int nextSequence() {
+ return nextSequence;
+ }
+
+ OptionalLong lastAckedOffset() {
+ if (lastAckedOffset != ProduceResponse.INVALID_OFFSET)
+ return OptionalLong.of(lastAckedOffset);
+ return OptionalLong.empty();
+ }
+
+ OptionalInt lastAckedSequence() {
+ if (lastAckedSequence !=
TxnPartitionEntry.NO_LAST_ACKED_SEQUENCE_NUMBER)
+ return OptionalInt.of(lastAckedSequence);
+ return OptionalInt.empty();
+ }
+
+ boolean hasInflightBatches() {
+ return !inflightBatchesBySequence.isEmpty();
+ }
+
+ ProducerBatch nextBatchBySequence() {
+ return inflightBatchesBySequence.isEmpty() ? null :
inflightBatchesBySequence.first();
+ }
+
+ void incrementSequence(int increment) {
+ this.nextSequence =
DefaultRecordBatch.incrementSequence(this.nextSequence, increment);
+ }
+
+ void addInflightBatch(ProducerBatch batch) {
+ inflightBatchesBySequence.add(batch);
+ }
+
+ void setLastAckedOffset(long lastAckedOffset) {
+ this.lastAckedOffset = lastAckedOffset;
+ }
+
+ void startSequencesAtBeginning(ProducerIdAndEpoch newProducerIdAndEpoch) {
+ final PrimitiveRef.IntRef sequence = PrimitiveRef.ofInt(0);
+ resetSequenceNumbers(inFlightBatch -> {
+ inFlightBatch.resetProducerState(newProducerIdAndEpoch,
sequence.value);
+ sequence.value += inFlightBatch.recordCount;
+ });
+ producerIdAndEpoch = newProducerIdAndEpoch;
+ nextSequence = sequence.value;
+ lastAckedSequence = NO_LAST_ACKED_SEQUENCE_NUMBER;
+ }
+
+ void adjustSequencesDueToFailedBatch(long baseSequence, int recordCount) {
+ decrementSequence(recordCount);
+ resetSequenceNumbers(inFlightBatch -> {
+ if (inFlightBatch.baseSequence() < baseSequence)
+ return;
+
+ int newSequence = inFlightBatch.baseSequence() - recordCount;
+ if (newSequence < 0)
+ throw new IllegalStateException("Sequence number for batch
with sequence " + inFlightBatch.baseSequence()
+ + " for partition " + topicPartition + " is going to
become negative: " + newSequence);
+
+ inFlightBatch.resetProducerState(new
ProducerIdAndEpoch(inFlightBatch.producerId(), inFlightBatch.producerEpoch()),
newSequence);
+ });
+ }
+
+ int maybeUpdateLastAckedSequence(int sequence) {
Review Comment:
nit: for these two methods: in TransactionManager, they were listed before
adjustSequencesDueToFailedBatch. Not a big deal, but was curious if we wanted
the ordering to be consistent.
--
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]
