You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ShivsundarR (via GitHub)" <gi...@apache.org> on 2025/03/26 18:28:38 UTC
[PR] MINOR: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
ShivsundarR opened a new pull request, #19295:
URL: https://github.com/apache/kafka/pull/19295
*What*
Currently if we received just a control record in the `ShareFetchResponse`, then the currentFetch in `ShareConsumerImpl` would not be updated as the record is ignored. But in the process, we lose the acknowledgment for this control record which is GAP for control records.
PR fixes this by adding a check to include the pending acknowledgements even when the `ShareFetch` is empty.
Added a unit test which verifies this change.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016349699
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
@chia7712 So we have ShareFetch and `ShareInFlightBatch`.
`ShareFetch` has a `Map<TopicIdPartition, ShareInFlightBatch<K, V>>` as a member.
`ShareFetch::isEmpty` does
```
public boolean isEmpty() {
return numRecords() == 0;
}
```
which iterates through the batches and checks ShareInFlightBatch::numRecords which is
```
int numRecords() {
return inFlightRecords.size();
}
```
The code you shared is from `ShareInFlightBatch::isEmpty()` where it checks both the `InFlightRecords` and `Acknowledgements`.
```
public boolean isEmpty() {
return inFlightRecords.isEmpty() && acknowledgements.isEmpty();
}
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
@chia7712 So we have `ShareFetch` and `ShareInFlightBatch`.
`ShareFetch` has a `Map<TopicIdPartition, ShareInFlightBatch<K, V>>` as a member.
`ShareFetch::isEmpty` does
```
public boolean isEmpty() {
return numRecords() == 0;
}
```
which iterates through the batches and checks ShareInFlightBatch::numRecords which is
```
int numRecords() {
return inFlightRecords.size();
}
```
The code you shared is from `ShareInFlightBatch::isEmpty()` where it checks both the `InFlightRecords` and `Acknowledgements`.
```
public boolean isEmpty() {
return inFlightRecords.isEmpty() && acknowledgements.isEmpty();
}
```
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016537417
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
> which iterates through the batches and checks ShareInFlightBatch::numRecords which is
I see. thanks for this clarification.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19295:
URL: https://github.com/apache/kafka/pull/19295#issuecomment-2769293324
Build failure due to excessively flaky test. Test disabled by https://github.com/apache/kafka/pull/19340.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are acknowledged with GAP (indicating the client is ignoring these control records) and are never presented to the consumer application.
- Now these control records are auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now for this PR, we have added any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` so that it can be sent on the next poll().
- I agree it looks a bit odd though for readability. But yeah there is a case when this could happen.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Kind of :)) So it would have empty records but could have non-empty acknowledgements (for skipped records).
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, so the records never arrive to the application thread, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now after this PR, any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` are added so that it can be sent on the next `poll()`.
- We cannot present these to the application, so the check for `fetch.isEmpty` cannot be altered. But yeah there is a case when this could happen.
- I agree it looks a bit odd though for readability. I will think of a different way of propagating the acks to the background thread.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now after this PR, any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` are added so that it can be sent on the next `poll()`.
- We cannot present these to the application, so the check for `fetch.isEmpty` cannot be altered. But yeah there is a case when this could happen. I agree it looks a bit odd though for readability.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on PR #19295:
URL: https://github.com/apache/kafka/pull/19295#issuecomment-2758130014
@AndrewJSchofield as @chia7712 mentioned if the case arises where successive `ShareFetchResponses` have only control records (eg. 2 abort markers received in read_commited mode),
and if the second response came in at `fetchBuffer.awaitNotEmpty(pollTimer);`, then when we do `collect(Collections.emptyMap())`, then we would get a new set of acknowledgements for control records, but this time, these incoming acknowledgements would be ignored and missed.
Here we might have to combine the acknowledgements.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now for this PR, we have added any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` so that it can be sent on the next poll().
- I agree it looks a bit odd though for readability. But yeah there is a case when this could happen.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016068726
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Excuse me, is it possible that `fetch.takeAcknowledgedRecords()` returns non empty records when `fetch.isEmpty` is true?
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016245476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
This is clearly not quite right yet. I want to see the log dump for the situation which causes this and understand the batching.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016581868
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
Review Comment:
Is it possible for two ShareFetchResponses to contain only control records for the same partition? If so, we should merge the AcknowledgedRecords for that specific partition
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016187929
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
@ShivsundarR thanks for explanation, and I have a follow-up question.
> But here as fetch.isEmpty() only checks for numRecords() == 0, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) -
`fetch.isEmpty() ` means all batches have no `inFlightRecords` and `acknowledgements`, right?
```java
public boolean isEmpty() {
return inFlightRecords.isEmpty() && acknowledgements.isEmpty();
}
```
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareInFlightBatch.java#L116
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2022174511
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java:
##########
@@ -1234,26 +1087,20 @@ public void testRetryAcknowledgementsWithLeaderChange() {
subscriptions.assignFromSubscribed(partitions);
client.updateMetadata(
- RequestTestUtils.metadataUpdateWithIds(2, Map.of(topicName, 1),
- tp -> validLeaderEpoch, topicIds, false));
+ RequestTestUtils.metadataUpdateWithIds(2, Map.of(topicName, 1),
Review Comment:
Thanks, yeah it was some IDE setting which changed the indentation automatically, I missed it.
I have changed this back now.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now for this PR, we have added any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` so that it can be sent on the next poll().
- I agree it looks a bit odd though for readability. But yeah there is a case when this could happen.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Kind of :)) So it would have empty records but could have non-empty acknowledgements (for skipped records).
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, so the records never arrive to the application thread, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now after this PR, any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` are added so that it can be sent on the next `poll()`.
- We cannot present these to the application, so the check for `fetch.isEmpty` cannot be altered. But yeah there is a case when this could happen. I agree it looks a bit odd though for readability.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are usually auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But here as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now after this PR, any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` are added so that it can be sent on the next poll().
- I agree it looks a bit odd though for readability. But yeah there is a case when this could happen.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2020709446
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java:
##########
@@ -1234,26 +1087,20 @@ public void testRetryAcknowledgementsWithLeaderChange() {
subscriptions.assignFromSubscribed(partitions);
client.updateMetadata(
- RequestTestUtils.metadataUpdateWithIds(2, Map.of(topicName, 1),
- tp -> validLeaderEpoch, topicIds, false));
+ RequestTestUtils.metadataUpdateWithIds(2, Map.of(topicName, 1),
Review Comment:
In quite a few places in this file, the indentation has been increased but the lines have not been changed apart from that. Please can you minimise these changes so the PR is much tighter.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java:
##########
@@ -2905,4 +2729,48 @@ public void add(BackgroundEvent event) {
}
}
}
+
+ @Test
+ void testFetchWithControlRecords() {
+ buildRequestManager();
+ shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsMap = new HashMap<>();
+
+ Acknowledgements acknowledgements = Acknowledgements.empty();
+ acknowledgements.add(1L, AcknowledgeType.ACCEPT);
+ nodeAcknowledgementsMap.put(tip0, new NodeAcknowledgements(0, acknowledgements));
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsControlRecordMap = new HashMap<>();
+
+ Acknowledgements controlAcknowledgements = Acknowledgements.empty();
+ controlAcknowledgements.addGap(2L);
+ nodeAcknowledgementsControlRecordMap.put(tip0, new NodeAcknowledgements(0, controlAcknowledgements));
+
+ shareConsumeRequestManager.fetch(nodeAcknowledgementsMap, nodeAcknowledgementsControlRecordMap);
+
+ Map<TopicIdPartition, Acknowledgements> fetchAcksToSend = shareConsumeRequestManager.getFetchAcknowledgementsToSend(0);
+ System.out.println(fetchAcksToSend);
+ assertEquals(1, fetchAcksToSend.size());
+ assertEquals(AcknowledgeType.ACCEPT, fetchAcksToSend.get(tip0).get(1L));
+ assertEquals(2, fetchAcksToSend.get(tip0).size());
+ assertNull(fetchAcksToSend.get(tip0).get(3L));
+ }
+
+ void sendFetchAndVerifyResponse(MemoryRecords records,
Review Comment:
This can be `private` I think.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java:
##########
@@ -2905,4 +2729,48 @@ public void add(BackgroundEvent event) {
}
}
}
+
+ @Test
+ void testFetchWithControlRecords() {
+ buildRequestManager();
+ shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsMap = new HashMap<>();
+
+ Acknowledgements acknowledgements = Acknowledgements.empty();
+ acknowledgements.add(1L, AcknowledgeType.ACCEPT);
+ nodeAcknowledgementsMap.put(tip0, new NodeAcknowledgements(0, acknowledgements));
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsControlRecordMap = new HashMap<>();
+
+ Acknowledgements controlAcknowledgements = Acknowledgements.empty();
+ controlAcknowledgements.addGap(2L);
+ nodeAcknowledgementsControlRecordMap.put(tip0, new NodeAcknowledgements(0, controlAcknowledgements));
+
+ shareConsumeRequestManager.fetch(nodeAcknowledgementsMap, nodeAcknowledgementsControlRecordMap);
+
+ Map<TopicIdPartition, Acknowledgements> fetchAcksToSend = shareConsumeRequestManager.getFetchAcknowledgementsToSend(0);
+ System.out.println(fetchAcksToSend);
+ assertEquals(1, fetchAcksToSend.size());
+ assertEquals(AcknowledgeType.ACCEPT, fetchAcksToSend.get(tip0).get(1L));
+ assertEquals(2, fetchAcksToSend.get(tip0).size());
+ assertNull(fetchAcksToSend.get(tip0).get(3L));
+ }
+
+ void sendFetchAndVerifyResponse(MemoryRecords records,
+ List<ShareFetchResponseData.AcquiredRecords> acquiredRecords,
+ Errors... error) {
Review Comment:
You are not using the varargs part of this. You could just use `Errors error`.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016129362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
Yes :)) turns out it can.
- Some integ tests in this PR - https://github.com/apache/kafka/pull/19261 revealed that in transactions, when client receives only a control record(eg. an abort marker) in the `ShareFetchResponse` (without any non-control record), then in the `ShareCompletedFetch`, these control records are never acknowledged(ideally acknowledged with GAP, indicating the client is ignoring these control records) and are never presented to the consumer application.
- It is expected that control records are skipped and are not presented to the application, but client should still acknowledge them with GAP (https://github.com/apache/kafka/blob/84b8fec089682486aa9827a3baffa2513118ce6d/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java#L33)
- Now these control records are auto acknowledged with `GAP` and will be sent on the next `ShareFetch`/`ShareAcknowledge` request. But as `fetch.isEmpty()` only checks for `numRecords() == 0`, when the fetch is empty, we actually ignore the fetch here(meaning we never acknowledge these control records) - https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java#L598
- Now for this PR, we have added any possible acknowledgements that came in with the empty fetch (from control records) to the `ShareFetchEvent` so that it can be sent on the next poll().
- I agree it looks a bit odd though for readability. But yeah there is a case when this could happen.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2022176927
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java:
##########
@@ -2905,4 +2729,48 @@ public void add(BackgroundEvent event) {
}
}
}
+
+ @Test
+ void testFetchWithControlRecords() {
+ buildRequestManager();
+ shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsMap = new HashMap<>();
+
+ Acknowledgements acknowledgements = Acknowledgements.empty();
+ acknowledgements.add(1L, AcknowledgeType.ACCEPT);
+ nodeAcknowledgementsMap.put(tip0, new NodeAcknowledgements(0, acknowledgements));
+
+ Map<TopicIdPartition, NodeAcknowledgements> nodeAcknowledgementsControlRecordMap = new HashMap<>();
+
+ Acknowledgements controlAcknowledgements = Acknowledgements.empty();
+ controlAcknowledgements.addGap(2L);
+ nodeAcknowledgementsControlRecordMap.put(tip0, new NodeAcknowledgements(0, controlAcknowledgements));
+
+ shareConsumeRequestManager.fetch(nodeAcknowledgementsMap, nodeAcknowledgementsControlRecordMap);
+
+ Map<TopicIdPartition, Acknowledgements> fetchAcksToSend = shareConsumeRequestManager.getFetchAcknowledgementsToSend(0);
+ System.out.println(fetchAcksToSend);
+ assertEquals(1, fetchAcksToSend.size());
+ assertEquals(AcknowledgeType.ACCEPT, fetchAcksToSend.get(tip0).get(1L));
+ assertEquals(2, fetchAcksToSend.get(tip0).size());
+ assertNull(fetchAcksToSend.get(tip0).get(3L));
+ }
+
+ void sendFetchAndVerifyResponse(MemoryRecords records,
+ List<ShareFetchResponseData.AcquiredRecords> acquiredRecords,
+ Errors... error) {
Review Comment:
There was one test which needed this, the change was missed when I pushed the commit. I have updated the PR with one usage of this varargs part. In future if we add tests with both top level error and acknowledge error, this might be useful as well.
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19295:
URL: https://github.com/apache/kafka/pull/19295
--
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: Add check in ShareConsumerImpl to send acknowledgements of control records when ShareFetch is empty. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #19295:
URL: https://github.com/apache/kafka/pull/19295#discussion_r2016349699
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -644,8 +645,12 @@ private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> ack
if (currentFetch.isEmpty()) {
final ShareFetch<K, V> fetch = fetchCollector.collect(fetchBuffer);
if (fetch.isEmpty()) {
+ // Check for any acknowledgements which could have come from control records (GAP) and include them.
+ Map<TopicIdPartition, NodeAcknowledgements> combinedAcknowledgements = new LinkedHashMap<>(acknowledgementsMap);
+ combinedAcknowledgements.putAll(fetch.takeAcknowledgedRecords());
+
// Fetch more records and send any waiting acknowledgements
- applicationEventHandler.add(new ShareFetchEvent(acknowledgementsMap));
+ applicationEventHandler.add(new ShareFetchEvent(combinedAcknowledgements));
Review Comment:
@chia7712 So we have ShareFetch and ShareInFlightBatch.
ShareFetch has a `Map<TopicIdPartition, ShareInFlightBatch<K, V>>` as an instance.
ShareFetch::isEmpty does
```
public boolean isEmpty() {
return numRecords() == 0;
}
```
which iterates through the batches and checks ShareInFlightBatch::numRecords which is
```
int numRecords() {
return inFlightRecords.size();
}
```
The code you shared is from ShareInFlightBatch::isEmpty() where it checks both the `InFlightRecords` and `Acknowledgements`.
```
public boolean isEmpty() {
return inFlightRecords.isEmpty() && acknowledgements.isEmpty();
}
```
--
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