This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1c573f84f6e0 fix(utilities): use endOffsets when no offset is greater
than the checkpoint timestamp (#19658)
1c573f84f6e0 is described below
commit 1c573f84f6e08d3942ad0dc1fe47cc6f1ea0eff7
Author: wangxianghu <[email protected]>
AuthorDate: Tue Aug 25 20:02:29 2026 +0400
fix(utilities): use endOffsets when no offset is greater than the
checkpoint timestamp (#19658)
* fix(utilities) Use endOffsets when there is no offsets greater than
timestamp checkpoint
* address comments
---
.../hudi/utilities/config/KafkaSourceConfig.java | 4 +
.../utilities/sources/helpers/KafkaOffsetGen.java | 40 ++++-
.../deltastreamer/TestHoodieDeltaStreamer.java | 68 +++++---
.../sources/helpers/TestKafkaOffsetGen.java | 177 +++++++++++++++++++++
4 files changed, 264 insertions(+), 25 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KafkaSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KafkaSourceConfig.java
index e0e06549ac02..d7b1dae8a663 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KafkaSourceConfig.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KafkaSourceConfig.java
@@ -58,6 +58,10 @@ public class KafkaSourceConfig extends HoodieConfig {
+ ". Default type is " + KAFKA_CHECKPOINT_TYPE_STRING + ". "
+ "For type " + KAFKA_CHECKPOINT_TYPE_STRING + ", checkpoint should
be provided as: topicName,0:offset0,1:offset1,2:offset2. "
+ "For type " + KAFKA_CHECKPOINT_TYPE_TIMESTAMP + ", checkpoint
should be provided as long value of desired timestamp. "
+ + "If a partition has no record with a timestamp at or after the
checkpoint (either the partition is empty, "
+ + "all records predate the checkpoint, or messages use a pre-0.10.0
format that has no timestamp), that "
+ + "partition resumes from its end offset (the tip) rather than from
the beginning; records already in "
+ + "that partition that predate the checkpoint are skipped and will
not be ingested. "
+ "For type " + KAFKA_CHECKPOINT_TYPE_SINGLE_OFFSET + ", we assume
that topic consists of a single partition, "
+ "so checkpoint should be provided as long value of desired
offset.");
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index 941858dc8552..aa76ac80842a 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -346,7 +346,7 @@ public class KafkaOffsetGen {
}
return CheckpointUtils.computeOffsetRanges(fromOffsets, toOffsets,
numEvents, minPartitions);
}
-
+
/**
* Fetch partition infos for given topic.
*
@@ -459,29 +459,59 @@ public class KafkaOffsetGen {
* 1. input: timestamp, etc.
* 2. output:
topicName,partition_num_0:100,partition_num_1:101,partition_num_2:102.
*
+ * <p>For each partition, {@link KafkaConsumer#offsetsForTimes} returns the
offset of the first
+ * record whose timestamp is >= {@code timestamp}. If a partition has no
such record (either
+ * because all records predate {@code timestamp}, the partition is empty, or
the messages use a
+ * pre-0.10.0 format that has no timestamp), {@code offsetsForTimes} returns
{@code null} for
+ * that partition. In that case, we fall back to the partition's end offset
rather than the
+ * beginning offset: since no record satisfies the requested timestamp, the
correct starting
+ * point is the tip of the partition. Falling back to the beginning offset
would re-consume the
+ * entire partition, which contradicts the semantics of a timestamp-based
checkpoint.
+ *
* @param consumer
* @param topicName
* @param timestamp
* @return
*/
- private Option<String> getOffsetsByTimestamp(KafkaConsumer consumer,
List<PartitionInfo> partitionInfoList, Set<TopicPartition> topicPartitions,
- String topicName, Long
timestamp) {
+ @VisibleForTesting
+ Option<String> getOffsetsByTimestamp(KafkaConsumer consumer,
List<PartitionInfo> partitionInfoList, Set<TopicPartition> topicPartitions,
+ String topicName, Long timestamp) {
Map<TopicPartition, Long> topicPartitionsTimestamp =
partitionInfoList.stream()
.map(x -> new
TopicPartition(x.topic(), x.partition()))
.collect(Collectors.toMap(Function.identity(), x -> timestamp));
- Map<TopicPartition, Long> earliestOffsets =
consumer.beginningOffsets(topicPartitions);
+ // Fetch end offsets BEFORE offsetsForTimes to close the window where a
record appended
+ // between the two calls would be permanently skipped. With this order, if
offsetsForTimes
+ // returns null for a partition, we fall back to the end offset captured
at T1; any record
+ // written after T1 will be picked up in the next batch (at-least-once
semantics).
+ Map<TopicPartition, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
+ // Track partitions with no offset at/after the requested timestamp so we
can surface them
+ // as a WARN. Without this, callers whose messages use a pre-0.10.0 format
(which always
+ // returns null here) would silently skip to the tip of every partition
with no signal.
+ Map<TopicPartition, Long> fallbackToEndOffsets = new HashMap<>();
StringBuilder sb = new StringBuilder(topicName);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> map :
offsetAndTimestamp.entrySet()) {
if (map.getValue() != null) {
sb.append(",").append(map.getKey().partition()).append(":").append(map.getValue().offset());
} else {
-
sb.append(",").append(map.getKey().partition()).append(":").append(earliestOffsets.get(map.getKey()));
+ // No record in this partition has a timestamp >= the requested one.
Fall back to the
+ // end offset captured before offsetsForTimes to guarantee
at-least-once: any record
+ // written after we snapshot endOffsets will be re-consumed next
batch, never skipped.
+ Long endOffset = endOffsets.get(map.getKey());
+ fallbackToEndOffsets.put(map.getKey(), endOffset);
+
sb.append(",").append(map.getKey().partition()).append(":").append(endOffset);
}
}
+ if (!fallbackToEndOffsets.isEmpty()) {
+ log.warn("No offset was found at/after timestamp {} for partitions;
falling back to their "
+ + "end offsets. This can happen when all records in the
partition predate the requested "
+ + "timestamp, when the partition is empty, or when messages use
a pre-0.10.0 format "
+ + "without a timestamp. Fallback offsets: {}",
+ timestamp, fallbackToEndOffsets);
+ }
return Option.of(sb.toString());
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index 16aeb35971e5..81c8560e61d9 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -3334,28 +3334,56 @@ public class TestHoodieDeltaStreamer extends
HoodieDeltaStreamerTestBase {
@Test
public void testKafkaTimestampType() throws Exception {
- topicName = "topic" + testNum;
+ // Timestamp-based Kafka checkpoints have two distinct fallback behaviors
we need to cover:
+ // (1) Checkpoint captured BEFORE records are produced: every record has
ts >= checkpoint,
+ // so `offsetsForTimes` returns concrete offsets and ingestion
consumes all of them.
+ // (2) Checkpoint captured AFTER records are produced: no record has ts
>= checkpoint, so
+ // `offsetsForTimes` returns null for every partition and we fall
back to the end offset
+ // of each partition. Nothing should be ingested, and a subsequent
batch produced *after*
+ // the checkpoint should be picked up on the next sync — this proves
that the fallback
+ // stored a usable checkpoint at the partition tip (not offset 0,
which would replay the
+ // original records).
kafkaCheckpointType = "timestamp";
- prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, topicName);
- prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName);
- String tableBasePath = basePath + "/test_json_kafka_table" + testNum;
- HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
- Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
- true, 100000, false, null,
- null, "timestamp", String.valueOf(System.currentTimeMillis())),
jsc);
- deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath, sqlContext);
- prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, false, topicName);
- deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
- Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
- true, 100000, false, null, null,
- "timestamp", String.valueOf(System.currentTimeMillis())), jsc);
- deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS * 2, tableBasePath, sqlContext);
- deltaStreamer.shutdownGracefully();
+ // ---- Case 1: checkpoint captured BEFORE producing records ----
+ long checkpointBeforeProduction = System.currentTimeMillis();
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, "topic" + testNum);
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
"topic" + testNum);
+ String tableBasePath1 = basePath + "/test_json_kafka_table" + testNum;
+ syncOnce(TestHelpers.makeConfig(tableBasePath1, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null,
+ null, "timestamp", String.valueOf(checkpointBeforeProduction)));
+ assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath1, sqlContext);
+
+ // ---- Case 2: checkpoint captured AFTER producing records ----
+ // First batch predates the checkpoint => fallback path returns end
offsets (partition tips).
+ // Nothing should be ingested in the first sync; a second batch produced
after the checkpoint
+ // should be fully consumed on the follow-up sync (which reuses the
checkpoint stored by the
+ // first sync). This asserts we resumed at the tip, not at offset 0.
+ String topicName2 = "topic_after_" + testNum;
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, topicName2);
+ // Small pause so the timestamp is guaranteed to be after the last
produced record's ts.
+ Thread.sleep(10);
+ long checkpointAfterProduction = System.currentTimeMillis();
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName2);
+ String tableBasePath2 = basePath + "/test_json_kafka_table_after_" +
testNum;
+ syncOnce(TestHelpers.makeConfig(tableBasePath2, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null, null,
+ "timestamp", String.valueOf(checkpointAfterProduction)));
+ assertRecordCount(0, tableBasePath2, sqlContext);
+
+ // Produce a fresh batch strictly after the checkpoint and sync again with
no --checkpoint
+ // override, so the streamer picks up from the offsets we stored in the
first sync.
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, false, topicName2);
+ syncOnce(TestHelpers.makeConfig(tableBasePath2, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null, null,
+ "timestamp", null));
+ // Only the second batch should be ingested; the first batch (which
predates the checkpoint)
+ // stays skipped, confirming the fallback resumed at the partition tip.
+ assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath2, sqlContext);
}
@Disabled("HUDI-6609")
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
index e630d6213974..7b261a44ddf7 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java
@@ -37,12 +37,17 @@ import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.streaming.kafka010.OffsetRange;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -186,6 +191,103 @@ public class TestKafkaOffsetGen {
assertEquals(500, nextOffsetRanges[0].untilOffset());
}
+ /**
+ * When the requested timestamp is later than every record in the topic,
+ * {@link org.apache.kafka.clients.consumer.KafkaConsumer#offsetsForTimes}
returns {@code null}
+ * for every partition. In that case we must fall back to the partition's
end offset (its tip),
+ * not to offset 0 / earliest — otherwise the entire partition would be
replayed even though the
+ * user asked for a strictly later checkpoint.
+ */
+ @Test
+ public void
testGetNextOffsetRangesFromTimestampCheckpointTypeWithNoOffsetsAfterTimestamp()
throws Exception {
+ HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator();
+ testUtils.createTopic(testTopicName, 1);
+ testUtils.sendMessages(testTopicName,
Helpers.jsonifyRecords(dataGenerator.generateInserts("000", 1000)));
+ // Ensure the checkpoint we pass is strictly after every published
record's timestamp.
+ Thread.sleep(10);
+ long checkpointAfterProduction = System.currentTimeMillis();
+
+ KafkaOffsetGen kafkaOffsetGen = new
KafkaOffsetGen(getConsumerConfigs("latest", KAFKA_CHECKPOINT_TYPE_TIMESTAMP));
+
+ OffsetRange[] nextOffsetRanges = kafkaOffsetGen.getNextOffsetRanges(
+ Option.of(new
StreamerCheckpointV2(String.valueOf(checkpointAfterProduction))), 500, metrics);
+ assertEquals(1, nextOffsetRanges.length);
+ // Fallback to end offset (the tip): from == until == 1000, nothing to
consume.
+ assertEquals(1000, nextOffsetRanges[0].fromOffset());
+ assertEquals(1000, nextOffsetRanges[0].untilOffset());
+ }
+
+ /**
+ * Mixed case: some partitions have records at/after the requested timestamp
and some don't.
+ * Only the partitions with no matching record should fall back to their end
offset; partitions
+ * that do have matching records should still resume at the offset returned
by
+ * {@link org.apache.kafka.clients.consumer.KafkaConsumer#offsetsForTimes}.
This is the user-visible
+ * bug the fallback change is targeting.
+ */
+ @Test
+ public void
testGetNextOffsetRangesFromTimestampCheckpointTypeWithPartialOffsets() throws
Exception {
+ testUtils.createTopic(testTopicName, 2);
+ int recordsPerPartition = 500;
+
+ // Publish `recordsPerPartition` records to partition 0 first, then take a
checkpoint after
+ // them. Any record produced later goes to partition 1 and is guaranteed
to have a timestamp
+ // strictly greater than the checkpoint.
+ sendMessagesToPartition(testTopicName, 0, recordsPerPartition);
+ Thread.sleep(10);
+ long checkpointBetweenBatches = System.currentTimeMillis();
+ Thread.sleep(10);
+ sendMessagesToPartition(testTopicName, 1, recordsPerPartition);
+
+ KafkaOffsetGen kafkaOffsetGen = new
KafkaOffsetGen(getConsumerConfigs("latest", KAFKA_CHECKPOINT_TYPE_TIMESTAMP));
+
+ OffsetRange[] nextOffsetRanges = kafkaOffsetGen.getNextOffsetRanges(
+ Option.of(new
StreamerCheckpointV2(String.valueOf(checkpointBetweenBatches))),
recordsPerPartition, metrics);
+
+ // computeOffsetRanges may split a single partition into multiple
sub-ranges when
+ // eventsPerPartition < partition size, so group by partition and verify
the aggregate.
+ Map<Integer, List<OffsetRange>> byPartition =
Arrays.stream(nextOffsetRanges)
+ .collect(Collectors.groupingBy(OffsetRange::partition));
+ assertEquals(2, byPartition.size(), "expected ranges for exactly 2
partitions");
+
+ // Partition 0: all records predate the checkpoint => fromOffset ==
untilOffset == 500
+ List<OffsetRange> p0Ranges = byPartition.get(0);
+ assertEquals(recordsPerPartition, p0Ranges.get(0).fromOffset(),
+ "partition 0 should start at the end offset (tip)");
+ assertEquals(recordsPerPartition, p0Ranges.get(p0Ranges.size() -
1).untilOffset(),
+ "partition 0 should end at the end offset (nothing to consume)");
+
+ // Partition 1: records were produced after the checkpoint => consume from
offset 0 to 500
+ List<OffsetRange> p1Ranges = byPartition.get(1);
+ assertEquals(0, p1Ranges.get(0).fromOffset(),
+ "partition 1 should start from offset 0");
+ assertEquals(recordsPerPartition, p1Ranges.get(p1Ranges.size() -
1).untilOffset(),
+ "partition 1 should consume all records");
+ assertEquals(recordsPerPartition,
KafkaOffsetGen.CheckpointUtils.totalNewMessages(nextOffsetRanges),
+ "total new messages should equal recordsPerPartition");
+ for (int i = 0; i < p1Ranges.size() - 1; i++) {
+ assertEquals(p1Ranges.get(i).untilOffset(), p1Ranges.get(i +
1).fromOffset(),
+ "partition 1 sub-ranges should be contiguous");
+ }
+ }
+
+ /**
+ * Publish {@code count} simple string records to a specific partition of
{@code topic}. Used to
+ * simulate a mixed "some partitions have records after ts, some don't"
state that the default
+ * partitioner-based {@link KafkaTestUtils#sendMessages} cannot
deterministically produce.
+ */
+ private void sendMessagesToPartition(String topic, int partition, int count)
{
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
testUtils.brokerAddress());
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerProps)) {
+ for (int i = 0; i < count; i++) {
+ producer.send(new ProducerRecord<>(topic, partition, null, "msg-" +
partition + "-" + i));
+ }
+ producer.flush();
+ }
+ }
+
@Test
public void testGetNextOffsetRangesFromSingleOffsetCheckpoint() {
HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator();
@@ -590,6 +692,81 @@ public class TestKafkaOffsetGen {
}
}
+ static Stream<Arguments> getOffsetsByTimestampArgs() {
+ long ts = System.currentTimeMillis();
+ String topicName = "kafka-topic-" + UUID.randomUUID();
+ List<TopicPartition> topicPartitions =
+ Arrays.asList(new TopicPartition(topicName, 0), new
TopicPartition(topicName, 1));
+
+ // end offsets used as fallback when offsetsForTimes returns null
+ Map<TopicPartition, Long> endOffsets = new HashMap<>();
+ endOffsets.put(topicPartitions.get(0), 50L);
+ endOffsets.put(topicPartitions.get(1), 80L);
+
+ // none-null: every partition resolves via offsetsForTimes
+ Map<TopicPartition, OffsetAndTimestamp> allResolved = new HashMap<>();
+ allResolved.put(topicPartitions.get(0), new OffsetAndTimestamp(20, ts));
+ allResolved.put(topicPartitions.get(1), new OffsetAndTimestamp(35, ts));
+ String expectedAllResolved = String.format("%s,0:20,1:35", topicName);
+
+ // some-null: one partition resolves, one falls back to end offset
+ Map<TopicPartition, OffsetAndTimestamp> someNull = new HashMap<>();
+ someNull.put(topicPartitions.get(0), new OffsetAndTimestamp(20, ts));
+ someNull.put(topicPartitions.get(1), null);
+ String expectedSomeNull = String.format("%s,0:20,1:80", topicName);
+
+ // all-null: every partition falls back to end offset (pre-0.10.0 format /
empty partitions)
+ Map<TopicPartition, OffsetAndTimestamp> allNull = new HashMap<>();
+ allNull.put(topicPartitions.get(0), null);
+ allNull.put(topicPartitions.get(1), null);
+ String expectedAllNull = String.format("%s,0:50,1:80", topicName);
+
+ return Stream.of(
+ Arguments.of(topicName, topicPartitions, endOffsets, allResolved,
expectedAllResolved),
+ Arguments.of(topicName, topicPartitions, endOffsets, someNull,
expectedSomeNull),
+ Arguments.of(topicName, topicPartitions, endOffsets, allNull,
expectedAllNull)
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("getOffsetsByTimestampArgs")
+ void testGetOffsetsByTimestamp(
+ String topicName,
+ List<TopicPartition> topicPartitions,
+ Map<TopicPartition, Long> endOffsets,
+ Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp,
+ String expectedCheckpoint) {
+ long timestamp = System.currentTimeMillis();
+
+ KafkaConsumer mockConsumer = mock(KafkaConsumer.class);
+ List<PartitionInfo> partitionInfoList = topicPartitions.stream()
+ .map(tp -> new org.apache.kafka.common.PartitionInfo(tp.topic(),
tp.partition(), null, null, null))
+ .collect(Collectors.toList());
+
+ Map<TopicPartition, Long> topicPartitionsTimestamp = new HashMap<>();
+ topicPartitions.forEach(tp -> topicPartitionsTimestamp.put(tp, timestamp));
+
+ when(mockConsumer.endOffsets(new
HashSet<>(topicPartitions))).thenReturn(endOffsets);
+
when(mockConsumer.offsetsForTimes(topicPartitionsTimestamp)).thenReturn(offsetAndTimestamp);
+
+ TypedProperties consumerConfigs = getConsumerConfigs(topicName,
"earliest", "string");
+ KafkaOffsetGen kafkaOffsetGen = new KafkaOffsetGen(consumerConfigs);
+
+ Option<String> result = kafkaOffsetGen.getOffsetsByTimestamp(
+ mockConsumer, partitionInfoList, new HashSet<>(topicPartitions),
topicName, timestamp);
+
+ assertTrue(result.isPresent());
+ // Parse both strings into offset maps for order-independent comparison
+ assertEquals(
+ KafkaOffsetGen.CheckpointUtils.strToOffsets(expectedCheckpoint),
+ KafkaOffsetGen.CheckpointUtils.strToOffsets(result.get()));
+
+ // endOffsets must be called BEFORE offsetsForTimes (verified via call
order)
+ org.mockito.InOrder inOrder = org.mockito.Mockito.inOrder(mockConsumer);
+ inOrder.verify(mockConsumer).endOffsets(new HashSet<>(topicPartitions));
+ inOrder.verify(mockConsumer).offsetsForTimes(topicPartitionsTimestamp);
+ }
+
void mockDescribeTopicConfigs(MockedStatic<AdminClient> staticMock, Map
kafkaParams, Config topicConfig) {
mockDescribeTopicConfigs(staticMock, kafkaParams, topicConfig,
testTopicName);
}
