You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ijuma (via GitHub)" <gi...@apache.org> on 2023/10/19 14:06:41 UTC
[PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
ijuma opened a new pull request, #14591:
URL: https://github.com/apache/kafka/pull/14591
And encapsulate TxnPartitionEntry state.
This makes it easier to understand the behavior and the paths through
which the state is updated.
### 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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #14591:
URL: https://github.com/apache/kafka/pull/14591#issuecomment-1774355395
@jolshan I think this is ready for review.
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1365609426
##########
clients/src/main/java/org/apache/kafka/common/utils/ProducerIdAndEpoch.java:
##########
@@ -35,7 +35,7 @@ public boolean isValid() {
@Override
public String toString() {
- return "ProducerIdAndEpoch(producerId=" + producerId + ", epoch=" + epoch + ")";
+ return "(producerId=" + producerId + ", epoch=" + epoch + ")";
Review Comment:
`ProducerIdAndEpoch` doesn't add much value here, this is effectively a tuple.
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375122091
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -778,22 +778,22 @@ public void testIdempotenceWithMultipleInflights() throws Exception {
prepareAndReceiveInitProducerId(producerId, Errors.NONE);
assertTrue(transactionManager.hasProducerId());
- assertEquals(0, transactionManager.sequenceNumber(tp0).longValue());
+ assertEquals(0, transactionManager.sequenceNumber(tp0));
Review Comment:
why were these longs in the first place?
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma merged PR #14591:
URL: https://github.com/apache/kafka/pull/14591
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #14591:
URL: https://github.com/apache/kafka/pull/14591#issuecomment-1783828937
JDK 8 build passed, the rest had a number of unrelated flaky failures:
> Build / JDK 21 and Scala 2.13 / kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest.testCreateTokenForOtherUserFails(String).quorum=kraft
Build / JDK 21 and Scala 2.13 / kafka.zk.ZkMigrationIntegrationTest.[2] Type=ZK, MetadataVersion=3.5-IV2, Security=PLAINTEXT
Build / JDK 21 and Scala 2.13 / org.apache.kafka.streams.integration.ConsistencyVectorIntegrationTest.shouldHaveSamePositionBoundActiveAndStandBy
Build / JDK 21 and Scala 2.13 / org.apache.kafka.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBootstrapZkMigrationRecord()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldAddNamedTopologyToRunningApplicationWithSingleInitialNamedTopology()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest.testSeparateOffsetsTopic
Build / JDK 11 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testNoOpRecordWriteAfterTimeout()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest.testSendOffsetsWithGroupId(String).quorum=zk
Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.ConsistencyVectorIntegrationTest.shouldHaveSamePositionBoundActiveAndStandBy
Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.EOSUncleanShutdownIntegrationTest.shouldWorkWithUncleanShutdownWipeOutStateStore[exactly_once_v2]
Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.EosIntegrationTest.shouldWriteLatestOffsetsToCheckpointOnShutdown[at_least_once, processing threads = false]
Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.EosIntegrationTest.shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once, processing threads = false]
Build / JDK 11 and Scala 2.13 / org.apache.kafka.tools.MetadataQuorumCommandTest.[6] Type=Raft-Isolated, Name=testDescribeQuorumStatusSuccessful, MetadataVersion=3.7-IV1, Security=PLAINTEXT
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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. (Looks like we are pretty consistent with the TxnPartitionMap ordering)
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375127610
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3449,7 +3449,8 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(new Metrics(t
transactionManager.handleCompletedBatch(tp1b3, t1b3Response);
assertFalse(transactionManager.hasInflightBatches(tp1));
- assertEquals(1, transactionManager.sequenceNumber(tp1).intValue());
+ assertEquals(1, transactionManager.sequenceNumber(tp1));
+ assertEquals(1, transactionManager.sequenceNumber(tp1));
Review Comment:
Good catch!
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375127506
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -778,22 +778,22 @@ public void testIdempotenceWithMultipleInflights() throws Exception {
prepareAndReceiveInitProducerId(producerId, Errors.NONE);
assertTrue(transactionManager.hasProducerId());
- assertEquals(0, transactionManager.sequenceNumber(tp0).longValue());
+ assertEquals(0, transactionManager.sequenceNumber(tp0));
Review Comment:
I can't think of a good reason.
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375128035
##########
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:
I tried to have non private methods first. No particular reason for these two, I moved them above `adjustSequencesDueToFailedBatch`.
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #14591:
URL: https://github.com/apache/kafka/pull/14591#discussion_r1375121838
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3449,7 +3449,8 @@ MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, new SenderMetricsRegistry(new Metrics(t
transactionManager.handleCompletedBatch(tp1b3, t1b3Response);
assertFalse(transactionManager.hasInflightBatches(tp1));
- assertEquals(1, transactionManager.sequenceNumber(tp1).intValue());
+ assertEquals(1, transactionManager.sequenceNumber(tp1));
+ assertEquals(1, transactionManager.sequenceNumber(tp1));
Review Comment:
nit: duplicated line?
--
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
Re: [PR] MINOR: Push down logic from TransactionManager to TxnPartitionEntry [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14591:
URL: https://github.com/apache/kafka/pull/14591#issuecomment-1780126154
Sorry I lost track of this. I will take a look soon.
--
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