You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2025/04/10 02:47:22 UTC
[PR] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
DL1231 opened a new pull request, #19430:
URL: https://github.com/apache/kafka/pull/19430
For records which are automatically released as a result of closing a share session normally, the delivery count should not be incremented. These records were fetched but they were not actually delivered to the client since the disposition of the delivery records is carried in the ShareAcknowledge which closes the share session. Any remaining records were not delivered, only fetched.
This PR releases the delivery count for records when closing a share session normally.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2055767896
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3003,24 +3014,32 @@ private boolean hasOngoingStateTransition() {
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise. Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
+ boolean decreaseFlag = false;
try {
+ if (ops == DeliveryCountOps.DECREASE && newState != RecordState.ARCHIVED) {
+ deliveryCount--;
Review Comment:
So will it be correct to change below check to following, and the later change the delivery count accordingly?
```
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
newState = RecordState.ARCHIVED;
}
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2043802656
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1011,6 +1011,8 @@ private Optional<Throwable> releaseAcquiredRecordsForPerOffsetBatch(String membe
return Optional.empty();
}
if (offsetState.getValue().state == RecordState.ACQUIRED) {
+ // These records were fetched but they were not actually delivered to the client.
+ offsetState.getValue().tryReleaseDeliveryCount();
Review Comment:
I considered to decrement the deliveryCount in `startStateTransition` and `startBatchStateTransition`. However, since `incrementDeliveryCount` is a boolean value that only controls whether to increment the count, introducing a decrementing functionality would require either adding an additional `decrementDeliveryCount` parameter or modifying `incrementDeliveryCount` to be an enum type.
What do you think?
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2059445472
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -4887,14 +4889,50 @@ public void testReleaseAcquiredRecordsBatchesPostStartOffsetMovementToMiddleOfBa
Map<Long, InFlightState> expectedOffsetStateMap = new HashMap<>();
expectedOffsetStateMap.put(10L, new InFlightState(RecordState.ARCHIVED, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(11L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(12L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(13L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(14L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(11L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(12L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(13L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(14L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
assertEquals(expectedOffsetStateMap, sharePartition.cachedState().get(10L).offsetState());
}
+ @Test
+ public void testReleaseAcquiredRecordsDecreaseDeliveryCount() {
Review Comment:
This test verifies that the delivery count is incremented before release and decremented after release
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2039566989
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1011,6 +1011,8 @@ private Optional<Throwable> releaseAcquiredRecordsForPerOffsetBatch(String membe
return Optional.empty();
}
if (offsetState.getValue().state == RecordState.ACQUIRED) {
+ // These records were fetched but they were not actually delivered to the client.
+ offsetState.getValue().tryReleaseDeliveryCount();
Review Comment:
So instead of creating another method to do this, should we enhance the API `startStateTransition` and `startBatchStateTransition` to decrement the deliveryCount on release?
I added the `incrementDeliveryCount` boolean to tackle the situation but seems we are missing the right behaviour while releasing. The state transition should be done within a transaction and not independently.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2048869119
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1011,6 +1011,8 @@ private Optional<Throwable> releaseAcquiredRecordsForPerOffsetBatch(String membe
return Optional.empty();
}
if (offsetState.getValue().state == RecordState.ACQUIRED) {
+ // These records were fetched but they were not actually delivered to the client.
+ offsetState.getValue().tryReleaseDeliveryCount();
Review Comment:
Sorry for the delay, I think either approach which you find sensible and readable should work. I might have preferred the enum based as that would be easy to read in future, tracking multiple booleans could be hard. But again, see how does it come after changes and choose either which you find sensible.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #19430:
URL: https://github.com/apache/kafka/pull/19430#issuecomment-2853819965
@apoorvmittal10 @AndrewJSchofield `group.share.delivery.count.limit` is limited to 2~10 and cannot be set to 1, do we need to modify this definition or modify this IT?
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2055365066
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3003,24 +3014,32 @@ private boolean hasOngoingStateTransition() {
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise. Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
+ boolean decreaseFlag = false;
try {
+ if (ops == DeliveryCountOps.DECREASE && newState != RecordState.ARCHIVED) {
+ deliveryCount--;
Review Comment:
Consider a situation where deliveryCount=4, deliveryCountLimit=5, and the message is delivered and the deliveryCount becomes 5.
The client is then shut down and needs to release the message, `ops=decrease`. If the operation is performed after `validateTransition`, it will first go to [here](https://github.com/apache/kafka/blob/71d08780d11b23ec4e931efaa8ca329c03f161e3/core/src/main/java/kafka/server/share/SharePartition.java#L3014), and because deliveryCount=5 at this point, newState is set to ARCHIVED.
In order to avoid this, the decrease of the deliveryCount is performed before `validateTransition`.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2052861184
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -197,6 +197,16 @@ public byte id() {
}
}
+ /**
+ * The DeliveryCountOps is used to specify the behavior on the delivery count: increase, decrease,
+ * or do nothing.
+ */
+ enum DeliveryCountOps {
+ INCREASE,
+ DECREASE,
+ NOOPS
Review Comment:
will it be better?
```suggestion
NO_OP
```
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3003,24 +3014,32 @@ private boolean hasOngoingStateTransition() {
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise. Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
+ boolean decreaseFlag = false;
try {
+ if (ops == DeliveryCountOps.DECREASE && newState != RecordState.ARCHIVED) {
+ deliveryCount--;
Review Comment:
This doesn't work as per the method docs i.e. `The delivery count is not incremented if the state update is unsuccessful.`
Shouldn't the change in deliveryCount happen after `validateTransition` check? Also can you please update the javadoc from `The delivery count is not incremented if the state update is unsuccessful.` => `The delivery count is not changed if the state update is unsuccessful.`
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3003,24 +3014,32 @@ private boolean hasOngoingStateTransition() {
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise. Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
+ boolean decreaseFlag = false;
try {
+ if (ops == DeliveryCountOps.DECREASE && newState != RecordState.ARCHIVED) {
+ deliveryCount--;
+ decreaseFlag = true;
+ }
if (newState == RecordState.AVAILABLE && deliveryCount >= maxDeliveryCount) {
newState = RecordState.ARCHIVED;
}
state = state.validateTransition(newState);
- if (incrementDeliveryCount && newState != RecordState.ARCHIVED) {
+ if (ops == DeliveryCountOps.INCREASE && newState != RecordState.ARCHIVED) {
deliveryCount++;
}
memberId = newMemberId;
return this;
} catch (IllegalStateException e) {
log.error("Failed to update state of the records", e);
+ if (decreaseFlag) {
+ deliveryCount++;
+ }
Review Comment:
We shouldn't need this change if we decrement the deliveryCount post stateTransition validation, correct?
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -4887,14 +4889,50 @@ public void testReleaseAcquiredRecordsBatchesPostStartOffsetMovementToMiddleOfBa
Map<Long, InFlightState> expectedOffsetStateMap = new HashMap<>();
expectedOffsetStateMap.put(10L, new InFlightState(RecordState.ARCHIVED, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(11L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(12L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(13L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
- expectedOffsetStateMap.put(14L, new InFlightState(RecordState.AVAILABLE, (short) 1, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(11L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(12L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(13L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
+ expectedOffsetStateMap.put(14L, new InFlightState(RecordState.AVAILABLE, (short) 0, EMPTY_MEMBER_ID));
assertEquals(expectedOffsetStateMap, sharePartition.cachedState().get(10L).offsetState());
}
+ @Test
+ public void testReleaseAcquiredRecordsDecreaseDeliveryCount() {
Review Comment:
How this test is different from earlier tests where delivery count was being decremented on release?
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -3652,7 +3652,8 @@ public void testReleaseSingleRecordBatch() {
assertEquals(0, sharePartition.nextFetchOffset());
assertEquals(1, sharePartition.cachedState().size());
assertEquals(RecordState.AVAILABLE, sharePartition.cachedState().get(0L).batchState());
- assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
Review Comment:
Can we have check `assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());` prio calling `releaseAcquiredRecords` which validates that deliveryCount was first incremented and then latter decremented?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -197,6 +197,16 @@ public byte id() {
}
}
+ /**
+ * The DeliveryCountOps is used to specify the behavior on the delivery count: increase, decrease,
+ * or do nothing.
+ */
+ enum DeliveryCountOps {
Review Comment:
```suggestion
private enum DeliveryCountOps {
```
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2066391545
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -2000,6 +1998,29 @@ public void testShareConsumerAfterCoordinatorMovement() throws Exception {
verifyShareGroupStateTopicRecordsProduced();
}
+ @ClusterTest
+ public void testDeliveryCountNotIncreaseAfterSessionClose() {
Review Comment:
Please can we have another test for explicit acknowledgement. In this case, for closing a share session normally, there is a difference between records which have been delivery and acknowledged by the application, and those which are released automatically as a result of the share session close. So, the test case could:
* Create a new share consumer setting explicit acknowledgement mode
* Fetch 2 records
* Acknowledge the first record with AcknowledgeType.RELEASE
* Commit the acknowledgement
* Close the consumer
* Create a new consumer and fetch the 2 records again
The delivery count for the first record should be 2, while the delivery count for the second record should be 1. This is because we know the first record was delivered, but the second we do not.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #19430:
URL: https://github.com/apache/kafka/pull/19430#issuecomment-2853825000
> @apoorvmittal10 @AndrewJSchofield `group.share.delivery.count.limit` is limited to 2~10 and cannot be set to 1, do we need to modify this definition or modify this IT?
Yeah we can set to 2 in IT.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2063589486
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2999,23 +3010,32 @@ private boolean hasOngoingStateTransition() {
/**
* Try to update the state of the records. The state of the records can only be updated if the
- * new state is allowed to be transitioned from old state. The delivery count is not incremented
+ * new state is allowed to be transitioned from old state. The delivery count is not changed
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise. Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
try {
- if (newState == RecordState.AVAILABLE && deliveryCount >= maxDeliveryCount) {
+ if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
newState = RecordState.ARCHIVED;
}
state = state.validateTransition(newState);
- if (incrementDeliveryCount && newState != RecordState.ARCHIVED) {
- deliveryCount++;
+ if (newState != RecordState.ARCHIVED) {
+ switch (ops) {
+ case INCREASE -> deliveryCount++;
+ case DECREASE -> deliveryCount--;
+ // do nothing
+ case NO_OP -> { }
+ default -> {
+ log.error("Invalid ops for delivery count.");
+ return null;
+ }
+ }
Review Comment:
nit: Is below cleaner? You won't require unneccessary `default` statement and method will be cleaner as well.
```
deliveryCount = updatedDeliveryCount(ops);
```
```
private int updatedDeliveryCount(DeliveryCountOps ops) {
return switch (ops) {
case INCREASE -> deliveryCount + 1;
case DECREASE -> deliveryCount - 1;
// do nothing
case NO_OP -> deliveryCount;
};
}
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19430:
URL: https://github.com/apache/kafka/pull/19430
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19430:
URL: https://github.com/apache/kafka/pull/19430#issuecomment-2794085366
@DL1231 Looks like a compile failure.
--
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] KAFKA-17541:[1/2] Improve handling of delivery count [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #19430:
URL: https://github.com/apache/kafka/pull/19430#issuecomment-2845094174
@AndrewJSchofield @apoorvmittal10 Thanks for your suggestion. I’ll finish it later.
--
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