[
https://issues.apache.org/jira/browse/KAFKA-7616?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16693581#comment-16693581
]
ASF GitHub Bot commented on KAFKA-7616:
---------------------------------------
omkreddy closed pull request #5901: KAFKA-7616: Make MockConsumer only add
entries to the partition map r…
URL: https://github.com/apache/kafka/pull/5901
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
index 9eee6da9d55..fba60344ba5 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
@@ -183,16 +183,13 @@ public synchronized void unsubscribe() {
// update the consumed offset
final Map<TopicPartition, List<ConsumerRecord<K, V>>> results = new
HashMap<>();
- for (final TopicPartition topicPartition : records.keySet()) {
- results.put(topicPartition, new ArrayList<>());
- }
for (Map.Entry<TopicPartition, List<ConsumerRecord<K, V>>> entry :
this.records.entrySet()) {
if (!subscriptions.isPaused(entry.getKey())) {
final List<ConsumerRecord<K, V>> recs = entry.getValue();
for (final ConsumerRecord<K, V> rec : recs) {
if (assignment().contains(entry.getKey()) && rec.offset()
>= subscriptions.position(entry.getKey())) {
- results.get(entry.getKey()).add(rec);
+ results.computeIfAbsent(entry.getKey(), partition ->
new ArrayList<>()).add(rec);
subscriptions.position(entry.getKey(), rec.offset() +
1);
}
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java
index 1d01eb6d0b2..03013e6a9fa 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java
@@ -26,8 +26,10 @@
import java.util.HashMap;
import java.util.Iterator;
+import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
public class MockConsumerTest {
@@ -84,4 +86,16 @@ public void testSimpleMockDeprecated() {
assertEquals(2L, consumer.committed(new TopicPartition("test",
0)).offset());
}
+ @Test
+ public void testConsumerRecordsIsEmptyWhenReturningNoRecords() {
+ TopicPartition partition = new TopicPartition("test", 0);
+ consumer.assign(Collections.singleton(partition));
+ consumer.addRecord(new ConsumerRecord<String, String>("test", 0, 0,
null, null));
+ consumer.updateEndOffsets(Collections.singletonMap(partition, 1L));
+ consumer.seekToEnd(Collections.singleton(partition));
+ ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1));
+ assertThat(records.count(), is(0));
+ assertThat(records.isEmpty(), is(true));
+ }
+
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> MockConsumer can return ConsumerRecords objects with a non-empty map but no
> records
> -----------------------------------------------------------------------------------
>
> Key: KAFKA-7616
> URL: https://issues.apache.org/jira/browse/KAFKA-7616
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 2.0.1
> Reporter: Stig Rohde Døssing
> Assignee: Stig Rohde Døssing
> Priority: Trivial
> Fix For: 2.2.0
>
>
> The ConsumerRecords returned from MockConsumer.poll can return false for
> isEmpty while not containing any records. This behavior is because
> MockConsumer.poll eagerly adds entries to the returned Map<TopicPartition,
> List<ConsumerRecord>>, based on which partitions have been added. If no
> records are returned for a partition, e.g. because the position was too far
> ahead, the entry for that partition will still be there.
>
> The MockConsumer should lazily add entries to the map as they are needed,
> since it is more in line with how the real consumer behaves.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
