You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/10/22 18:42:17 UTC
(kafka) branch trunk updated: KAFKA-17686 AsyncKafkaConsumer.offsetsForTimes() fails with NullPointerException (#17353)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d36eb9ed3f0 KAFKA-17686 AsyncKafkaConsumer.offsetsForTimes() fails with NullPointerException (#17353)
d36eb9ed3f0 is described below
commit d36eb9ed3f0ba3d9b58671f7bd74f82b7bbf6b9e
Author: Kirk True <ki...@kirktrue.pro>
AuthorDate: Tue Oct 22 11:42:10 2024 -0700
KAFKA-17686 AsyncKafkaConsumer.offsetsForTimes() fails with NullPointerException (#17353)
The code to convert the Map was initially expressed using the Streams API with collect(Collectors.toMap()). Unfortunately, the implementation of Collector returned by Collectors.toMap() does not support null entries. The KafkaConsumer.offsetsForTimes() API explicitly states that nulls can be present in the returned Map, hence this change.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../consumer/internals/AsyncKafkaConsumer.java | 11 ++++----
.../kafka/api/PlaintextAdminIntegrationTest.scala | 32 ++++++++++++++++------
2 files changed, 28 insertions(+), 15 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 1e4d0cf195b..35eccc99e12 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -100,6 +100,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -1068,12 +1069,10 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
}
try {
- return applicationEventHandler.addAndGet(listOffsetsEvent)
- .entrySet()
- .stream()
- .collect(Collectors.toMap(
- Map.Entry::getKey,
- entry -> entry.getValue().buildOffsetAndTimestamp()));
+ Map<TopicPartition, OffsetAndTimestampInternal> offsets = applicationEventHandler.addAndGet(listOffsetsEvent);
+ Map<TopicPartition, OffsetAndTimestamp> results = new HashMap<>(offsets.size());
+ offsets.forEach((k, v) -> results.put(k, v != null ? v.buildOffsetAndTimestamp() : null));
+ return results;
} catch (TimeoutException e) {
throw new TimeoutException("Failed to get offsets by times in " + timeout.toMillis() + "ms");
}
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 7336f96620e..7c429283179 100644
--- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -29,7 +29,7 @@ import kafka.integration.KafkaServerTestHarness
import kafka.server.metadata.KRaftMetadataCache
import kafka.server.KafkaConfig
import kafka.utils.TestUtils._
-import kafka.utils.{Log4jController, TestUtils}
+import kafka.utils.{Log4jController, TestInfoUtils, TestUtils}
import org.apache.kafka.clients.HostResolver
import org.apache.kafka.clients.admin.ConfigEntry.ConfigSource
import org.apache.kafka.clients.admin._
@@ -56,7 +56,7 @@ import org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Disabled, TestInfo, Timeout}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
+import org.junit.jupiter.params.provider.{Arguments, MethodSource, ValueSource}
import org.slf4j.LoggerFactory
import java.util.AbstractMap.SimpleImmutableEntry
@@ -1512,27 +1512,34 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertEquals(expectedLEO, brokers.head.replicaManager.localLog(topicPartition).get.logEndOffset)
}
- @ParameterizedTest
- @ValueSource(strings = Array("kraft"))
- def testOffsetsForTimesAfterDeleteRecords(quorum: String): Unit = {
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testOffsetsForTimesAfterDeleteRecords(quorum: String, groupProtocol: String): Unit = {
createTopic(topic, numPartitions = 2, replicationFactor = brokerCount)
client = createAdminClient
- val consumer = createConsumer()
+ val props = new Properties()
+ val consumer = createConsumer(configOverrides = props)
subscribeAndWaitForAssignment(topic, consumer)
val producer = createProducer()
sendRecords(producer, 10, topicPartition)
- assertEquals(0L, consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition).offset())
+ var returnedOffsets = consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava)
+ assertTrue(returnedOffsets.containsKey(topicPartition))
+ assertEquals(0L, returnedOffsets.get(topicPartition).offset())
var result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(5L)).asJava)
result.all.get
- assertEquals(5L, consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition).offset())
+ returnedOffsets = consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava)
+ assertTrue(returnedOffsets.containsKey(topicPartition))
+ assertEquals(5L, returnedOffsets.get(topicPartition).offset())
result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava)
result.all.get
- assertNull(consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition))
+ returnedOffsets = consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava)
+ assertTrue(returnedOffsets.containsKey(topicPartition))
+ assertNull(returnedOffsets.get(topicPartition))
}
@ParameterizedTest
@@ -3593,4 +3600,11 @@ object PlaintextAdminIntegrationTest {
assertEquals(LogConfig.DEFAULT_COMPRESSION_TYPE, configs.get(brokerResource).get(ServerConfigs.COMPRESSION_TYPE_CONFIG).value)
}
+
+ def getTestQuorumAndGroupProtocolParametersAll() : java.util.stream.Stream[Arguments] = {
+ util.Arrays.stream(Array(
+ Arguments.of("kraft", "classic"),
+ Arguments.of("kraft", "consumer")
+ ))
+ }
}