AndrewJSchofield commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4046068353
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -156,11 +160,15 @@ public void takeRenewedRecords() {
* @param type The acknowledge type which indicates whether it was
processed successfully
*/
public void acknowledge(final ConsumerRecord<K, V> record, final
AcknowledgeType type) {
- for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> tipBatch :
batches.entrySet()) {
- TopicIdPartition tip = tipBatch.getKey();
+ for (Map.Entry<TopicIdPartition, List<ShareInFlightBatch<K, V>>> entry
: batches.entrySet()) {
+ TopicIdPartition tip = entry.getKey();
if (tip.topic().equals(record.topic()) && (tip.partition() ==
record.partition())) {
- tipBatch.getValue().acknowledge(record, type);
- return;
+ for (ShareInFlightBatch<K, V> batch : entry.getValue()) {
+ if (batch.isInFlight(record.offset())) {
Review Comment:
I think what used to happen was that the records were merged, and the
`checkForRenewAcknowledgements` was copied into the merged batch too. Now, with
this PR, both batches maintain their own `checkForRenewAcknowledgements`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
