AndrewJSchofield commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4061207605
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -81,7 +78,13 @@ public void add(TopicIdPartition partition,
ShareInFlightBatch<K, V> batch) {
*/
public Map<TopicPartition, List<ConsumerRecord<K, V>>> records() {
final LinkedHashMap<TopicPartition, List<ConsumerRecord<K, V>>> result
= new LinkedHashMap<>();
- batches.forEach((tip, batch) -> result.put(tip.topicPartition(),
batch.getInFlightRecords()));
+ batches.forEach((tip, batchList) -> {
+ List<ConsumerRecord<K, V>> records = new ArrayList<>();
+ for (ShareInFlightBatch<K, V> batch : batchList) {
+ records.addAll(batch.getInFlightRecords());
+ }
+ result.put(tip.topicPartition(), records);
Review Comment:
We can omit adding an empty list of records, for the special case in which
the `ShareInFlightBatch` only contained control records. This would just mean
that the application would not see an empty list of `ConsumerRecord` for some
topic-partition, which could in principle occur today. We do need to be
super-careful not to confuse a batch with no non-control records with a batch
with no useful content (acks). This was the crux of the mistake which lead to
this issue to start with.
--
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]
