You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/06/15 14:41:59 UTC
[GitHub] [kafka] dajac commented on a change in pull request #10760: KAFKA-12541 Extend ListOffset to fetch offset with max timestamp
dajac commented on a change in pull request #10760:
URL: https://github.com/apache/kafka/pull/10760#discussion_r651831839
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -4298,6 +4314,18 @@ void handleFailure(Throwable throwable) {
}
}
}
+
+ @Override
+ boolean handleUnsupportedVersionException(UnsupportedVersionException exception) {
+ if (supportsMaxTimestamp) {
+ supportsMaxTimestamp = false;
+ // check if there are any non MAX_TIMESTAMPS partitions left to be downgraded
+ return partitionsToQuery.stream().anyMatch(
+ t -> t.partitions().stream().anyMatch(
+ p -> p.timestamp() != ListOffsetsRequest.MAX_TIMESTAMP));
+ }
Review comment:
Would it make sense to directly fail the future of the "max timestamp" requests here and to directly `partitionsToQuery` to only contains the remaining types? That would consolidate all the fall back logic here which is simpler to follow. Then, we could add another boolean `requireMaxTimestamp` to `ListOffsetsRequest.Builder.forConsumer` and we could directly pass `supportsMaxTimestamp` to it. This way, `createRequest` and `handleResponse` would remain unchanged. Did you already consider this?
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -4226,6 +4227,137 @@ public void testListOffsetsNonRetriableErrors() throws Exception {
}
}
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedSingleOffsetSpec() {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ Collections.singleton(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node})),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster, AdminClientConfig.RETRIES_CONFIG, "2")) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(Collections.singletonMap(tp0, OffsetSpec.maxTimestamp()));
+
+ Exception exception = assertThrows(ExecutionException.class, () -> {
+ Map<TopicPartition, ListOffsetsResultInfo> offsets = result.all().get(3, TimeUnit.SECONDS);
+ });
+ assertTrue(exception.getCause() instanceof UnsupportedVersionException);
+ }
+ }
+
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedMultipleOffsetSpec() throws Exception {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ List<PartitionInfo> pInfos = new ArrayList<>();
+ pInfos.add(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node}));
+ pInfos.add(new PartitionInfo("foo", 1, node, new Node[]{node}, new Node[]{node}));
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ pInfos,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+ final TopicPartition tp1 = new TopicPartition("foo", 1);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster,
+ AdminClientConfig.RETRIES_CONFIG, "2")) {
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(new HashMap<TopicPartition, OffsetSpec>() {{
+ put(tp0, OffsetSpec.maxTimestamp());
+ put(tp1, OffsetSpec.latest());
+ }});
+
+ ListOffsetsTopicResponse topicResponse = ListOffsetsResponse.singletonListOffsetsTopicResponse(tp1, Errors.NONE, -1L, 345L, 543);
+ ListOffsetsResponseData responseData = new ListOffsetsResponseData()
+ .setThrottleTimeMs(0)
+ .setTopics(Arrays.asList(topicResponse));
+ env.kafkaClient().prepareResponseFrom(new ListOffsetsResponse(responseData), node);
+
+ Exception exception = assertThrows(ExecutionException.class, () -> {
+ result.partitionResult(tp0).get();
+ });
+ assertTrue(exception.getCause() instanceof UnsupportedVersionException);
+
+ ListOffsetsResultInfo tp1Offset = result.partitionResult(tp1).get();
+ assertEquals(345L, tp1Offset.offset());
+ assertEquals(543, tp1Offset.leaderEpoch().get().intValue());
+ assertEquals(-1L, tp1Offset.timestamp());
+ }
+ }
+
+ @Test
+ public void testListOffsetsMaxTimestampAndNoBrokerResponse() {
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ List<PartitionInfo> pInfos = new ArrayList<>();
+ pInfos.add(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node}));
+ pInfos.add(new PartitionInfo("foo", 1, node, new Node[]{node}, new Node[]{node}));
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ pInfos,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+ final TopicPartition tp1 = new TopicPartition("foo", 1);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster,
+ AdminClientConfig.RETRIES_CONFIG, "2")) {
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(new HashMap<TopicPartition, OffsetSpec>() {{
+ put(tp0, OffsetSpec.maxTimestamp());
+ put(tp1, OffsetSpec.latest());
+ }});
+
+ ListOffsetsResponseData responseData = new ListOffsetsResponseData()
+ .setThrottleTimeMs(0)
+ .setTopics(new ArrayList<>());
+ env.kafkaClient().prepareResponseFrom(new ListOffsetsResponse(responseData), node);
+
+ Exception maxTimestampException = assertThrows(ExecutionException.class, () -> {
+ result.partitionResult(tp0).get();
+ });
+ assertTrue(maxTimestampException.getCause() instanceof UnsupportedVersionException);
+
+ Exception nopResponseException = assertThrows(ExecutionException.class, () -> {
+ result.partitionResult(tp1).get();
+ });
+ assertTrue(nopResponseException.getCause() instanceof ApiException);
+ }
+ }
Review comment:
Could we also add a unit test for the happy path with max timestamp?
##########
File path: core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package integration.kafka.admin
+
+import kafka.integration.KafkaServerTestHarness
+import kafka.server.KafkaConfig
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin._
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.utils.Utils
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
+
+import scala.collection.{Map, Seq}
+import scala.jdk.CollectionConverters._
+
+class ListOffsetsIntegrationTest extends KafkaServerTestHarness {
+
+ val topicName = "foo"
+ var adminClient: Admin = null
+
+ @BeforeEach
+ override def setUp(): Unit = {
+ super.setUp()
+ createTopic(topicName, 1, 1.asInstanceOf[Short])
Review comment:
nit: You can use `.toShort`.
##########
File path: core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package integration.kafka.admin
+
+import kafka.integration.KafkaServerTestHarness
+import kafka.server.KafkaConfig
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin._
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.utils.Utils
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
+
+import scala.collection.{Map, Seq}
+import scala.jdk.CollectionConverters._
+
+class ListOffsetsIntegrationTest extends KafkaServerTestHarness {
+
+ val topicName = "foo"
+ var adminClient: Admin = null
+
+ @BeforeEach
+ override def setUp(): Unit = {
+ super.setUp()
+ createTopic(topicName,1,1.asInstanceOf[Short])
+ produceMessages()
+ adminClient = Admin.create(Map[String, Object](
+ AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> brokerList
+ ).asJava)
+ }
+
+ @AfterEach
+ override def tearDown(): Unit = {
+ Utils.closeQuietly(adminClient, "ListOffsetsAdminClient")
+ super.tearDown()
+ }
+
+ @Test
+ def testEarliestOffset(): Unit = {
+ val earliestOffset = runFetchOffsets(adminClient, OffsetSpec.earliest())
+ assertEquals(0,earliestOffset.offset())
+ }
+
+ @Test
+ def testLatestOffset(): Unit = {
+ val latestOffset = runFetchOffsets(adminClient, OffsetSpec.latest())
+ assertEquals(3,latestOffset.offset())
+ }
+
+ @Test
+ def testMaxTimestampOffset(): Unit = {
+ val maxTimestampOffset = runFetchOffsets(adminClient, OffsetSpec.maxTimestamp())
+ assertEquals(1,maxTimestampOffset.offset())
+ }
+
+ private def runFetchOffsets(adminClient: Admin,
+ offsetSpec: OffsetSpec): ListOffsetsResult.ListOffsetsResultInfo = {
+ println(s"==> listOffsets(${topicName} -> ${offsetSpec}, new ListOffsetsOptions())")
Review comment:
Yeah, let's remove it.
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -4226,6 +4227,137 @@ public void testListOffsetsNonRetriableErrors() throws Exception {
}
}
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedSingleOffsetSpec() {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ Collections.singleton(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node})),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster, AdminClientConfig.RETRIES_CONFIG, "2")) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(Collections.singletonMap(tp0, OffsetSpec.maxTimestamp()));
+
+ Exception exception = assertThrows(ExecutionException.class, () -> {
+ Map<TopicPartition, ListOffsetsResultInfo> offsets = result.all().get(3, TimeUnit.SECONDS);
+ });
+ assertTrue(exception.getCause() instanceof UnsupportedVersionException);
+ }
+ }
+
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedMultipleOffsetSpec() throws Exception {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ List<PartitionInfo> pInfos = new ArrayList<>();
+ pInfos.add(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node}));
+ pInfos.add(new PartitionInfo("foo", 1, node, new Node[]{node}, new Node[]{node}));
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ pInfos,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+ final TopicPartition tp1 = new TopicPartition("foo", 1);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster,
+ AdminClientConfig.RETRIES_CONFIG, "2")) {
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(new HashMap<TopicPartition, OffsetSpec>() {{
+ put(tp0, OffsetSpec.maxTimestamp());
+ put(tp1, OffsetSpec.latest());
+ }});
Review comment:
nit: Could we indent the block such that `}});` is aligned with `ListOffsetsResult`? Same for other tests.
##########
File path: core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
##########
@@ -92,6 +92,57 @@ class LogOffsetTest extends BaseRequestTest {
assertEquals(Seq(20L, 18L, 16L, 14L, 12L, 10L, 8L, 6L, 4L, 3L), consumerOffsets)
}
+ @Test
+ def testFetchOffsetByTimestampForMaxTimestampAfterTruncate(): Unit = {
+ val topic = "kafka-"
+ val topicPartition = new TopicPartition(topic, 0)
+
+ createTopic(topic, 1, 1)
+
+ val logManager = server.getLogManager
+ TestUtils.waitUntilTrue(() => logManager.getLog(topicPartition).isDefined,
+ "Log for partition [topic,0] should be created")
+ val log = logManager.getLog(topicPartition).get
Review comment:
This block is common in all the test cases that have been added. Could we extract it into a helper method?
##########
File path: core/src/test/scala/unit/kafka/server/ListOffsetsRequestTest.scala
##########
@@ -162,11 +162,15 @@ class ListOffsetsRequestTest extends BaseRequestTest {
val partitionToLeader = TestUtils.createTopic(zkClient, topic, numPartitions = 1, replicationFactor = 3, servers)
val firstLeaderId = partitionToLeader(partition.partition)
- TestUtils.generateAndProduceMessages(servers, topic, 10)
+ // produce in 2 batches to ensure the max timestamp matches the last message
+ TestUtils.generateAndProduceMessages(servers, topic, 9)
+ Thread.sleep(10)
+ TestUtils.generateAndProduceMessages(servers, topic, 1)
Review comment:
It this really necessary? If yes, could we remove the `sleep`?
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -4226,6 +4227,137 @@ public void testListOffsetsNonRetriableErrors() throws Exception {
}
}
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedSingleOffsetSpec() {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ Collections.singleton(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node})),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster, AdminClientConfig.RETRIES_CONFIG, "2")) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(Collections.singletonMap(tp0, OffsetSpec.maxTimestamp()));
+
+ Exception exception = assertThrows(ExecutionException.class, () -> {
+ Map<TopicPartition, ListOffsetsResultInfo> offsets = result.all().get(3, TimeUnit.SECONDS);
+ });
+ assertTrue(exception.getCause() instanceof UnsupportedVersionException);
+ }
+ }
+
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedMultipleOffsetSpec() throws Exception {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ List<PartitionInfo> pInfos = new ArrayList<>();
+ pInfos.add(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node}));
+ pInfos.add(new PartitionInfo("foo", 1, node, new Node[]{node}, new Node[]{node}));
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ pInfos,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+ final TopicPartition tp1 = new TopicPartition("foo", 1);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster,
+ AdminClientConfig.RETRIES_CONFIG, "2")) {
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(new HashMap<TopicPartition, OffsetSpec>() {{
+ put(tp0, OffsetSpec.maxTimestamp());
+ put(tp1, OffsetSpec.latest());
+ }});
+
+ ListOffsetsTopicResponse topicResponse = ListOffsetsResponse.singletonListOffsetsTopicResponse(tp1, Errors.NONE, -1L, 345L, 543);
+ ListOffsetsResponseData responseData = new ListOffsetsResponseData()
+ .setThrottleTimeMs(0)
+ .setTopics(Arrays.asList(topicResponse));
+ env.kafkaClient().prepareResponseFrom(new ListOffsetsResponse(responseData), node);
Review comment:
nit: I would move this block up, before calling `listOffsets`, in order to have the response ready. Same for other tests.
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -4226,6 +4227,137 @@ public void testListOffsetsNonRetriableErrors() throws Exception {
}
}
+ @Test
+ public void testListOffsetsMaxTimestampUnsupportedSingleOffsetSpec() {
+
+ Node node = new Node(0, "localhost", 8120);
+ List<Node> nodes = Collections.singletonList(node);
+ final Cluster cluster = new Cluster(
+ "mockClusterId",
+ nodes,
+ Collections.singleton(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node})),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ node);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster, AdminClientConfig.RETRIES_CONFIG, "2")) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+
+ // listoffsets response from broker 0
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ request -> request instanceof ListOffsetsRequest);
+
+ ListOffsetsResult result = env.adminClient().listOffsets(Collections.singletonMap(tp0, OffsetSpec.maxTimestamp()));
+
+ Exception exception = assertThrows(ExecutionException.class, () -> {
+ Map<TopicPartition, ListOffsetsResultInfo> offsets = result.all().get(3, TimeUnit.SECONDS);
+ });
+ assertTrue(exception.getCause() instanceof UnsupportedVersionException);
Review comment:
nit: We could use `TestUtils.assertFutureThrows` here.
##########
File path: core/src/main/scala/kafka/log/Log.scala
##########
@@ -1316,6 +1316,16 @@ class Log(@volatile private var _dir: File,
val latestEpochOpt = leaderEpochCache.flatMap(_.latestEpoch).map(_.asInstanceOf[Integer])
val epochOptional = Optional.ofNullable(latestEpochOpt.orNull)
Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, logEndOffset, epochOptional))
+ } else if (targetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
+ // Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
+ // constant time access while being safe to use with concurrent collections unlike `toArray`.
+ val segmentsCopy = logSegments.toBuffer
+ val latestTimestampSegment = segmentsCopy.maxBy(_.maxTimestampSoFar)
+ val latestEpochOpt = leaderEpochCache.flatMap(_.latestEpoch).map(_.asInstanceOf[Integer])
+ val epochOptional = Optional.ofNullable(latestEpochOpt.orNull)
+ Some(new TimestampAndOffset(latestTimestampSegment.maxTimestampSoFar,
+ latestTimestampSegment.offsetOfMaxTimestampSoFar,
+ epochOptional))
Review comment:
I think that we should address this as well. I am fine with doing it in a follow-up PR though so we can keep this focused. Ok for you?
##########
File path: core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
##########
@@ -92,6 +92,57 @@ class LogOffsetTest extends BaseRequestTest {
assertEquals(Seq(20L, 18L, 16L, 14L, 12L, 10L, 8L, 6L, 4L, 3L), consumerOffsets)
}
+ @Test
+ def testFetchOffsetByTimestampForMaxTimestampAfterTruncate(): Unit = {
+ val topic = "kafka-"
+ val topicPartition = new TopicPartition(topic, 0)
+
+ createTopic(topic, 1, 1)
+
+ val logManager = server.getLogManager
+ TestUtils.waitUntilTrue(() => logManager.getLog(topicPartition).isDefined,
+ "Log for partition [topic,0] should be created")
+ val log = logManager.getLog(topicPartition).get
+
+ for (timestamp <- 0 until 20)
+ log.appendAsLeader(TestUtils.singletonRecords(value = Integer.toString(42).getBytes(), timestamp = timestamp.toLong), leaderEpoch = 0)
+ log.flush()
+
+ log.updateHighWatermark(log.logEndOffset)
+
+ val firstOffset = log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP)
+ assertEquals(19L, firstOffset.get.offset)
+
+ log.truncateTo(0)
+
+ val secondOffset = log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP)
+ assertEquals(0L, secondOffset.get.offset)
+
+ }
+
+ @Test
+ def testFetchOffsetByTimestampForMaxTimestampWithUnorderedTimestamps(): Unit = {
+ val topic = "kafka-"
+ val topicPartition = new TopicPartition(topic, 0)
+
+ createTopic(topic, 1, 1)
+
+ val logManager = server.getLogManager
+ TestUtils.waitUntilTrue(() => logManager.getLog(topicPartition).isDefined,
+ "Log for partition [topic,0] should be created")
+ val log = logManager.getLog(topicPartition).get
+
+ for (timestamp <- List(0L, 1L, 2L, 3L, 4L, 6L, 5L))
+ log.appendAsLeader(TestUtils.singletonRecords(value = Integer.toString(42).getBytes(), timestamp = timestamp), leaderEpoch = 0)
+ log.flush()
+
+ log.updateHighWatermark(log.logEndOffset)
+
+ val maxTimestampOffset = log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP)
+ assertEquals(7L, log.logEndOffset)
+ assertEquals(5L, maxTimestampOffset.get.offset)
Review comment:
Should we check the timestamp as well here and in the other tests as well?
##########
File path: core/src/test/scala/unit/kafka/log/LogTest.scala
##########
@@ -2072,6 +2072,30 @@ class LogTest {
log.fetchOffsetByTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP))
}
+ @Test
+ def testFetchOffsetByTimestampWithMaxTimestampIncludesTimestamp(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1)
+ val log = createLog(logDir, logConfig)
+
+ assertEquals(None, log.fetchOffsetByTimestamp(0L))
+
+ val firstTimestamp = mockTime.milliseconds
+ val leaderEpoch = 0
+ log.appendAsLeader(TestUtils.singletonRecords(
+ value = TestUtils.randomBytes(10),
+ timestamp = firstTimestamp),
+ leaderEpoch = leaderEpoch)
+
+ val secondTimestamp = firstTimestamp + 1
Review comment:
Should we produce a third record with a timestamp lower than this one to ensure that the API returns the maximum one and not the latest?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org