voonhous commented on code in PR #19658:
URL: https://github.com/apache/hudi/pull/19658#discussion_r3830459618
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,17 +480,39 @@ private Option<String>
getOffsetsByTimestamp(KafkaConsumer consumer, List<Partit
.map(x -> new
TopicPartition(x.topic(), x.partition()))
.collect(Collectors.toMap(Function.identity(), x -> timestamp));
- Map<TopicPartition, Long> earliestOffsets =
consumer.beginningOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
+ // Fetch end offsets lazily: they are only needed when at least one
partition has no record
+ // at/after the requested timestamp. In the common case where every
partition resolves,
+ // this saves one round-trip to the broker on every sync.
+ Map<TopicPartition, Long> endOffsets = null;
+ // 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 (partition tip) rather than 0/earliest to avoid
re-consuming the whole
+ // partition when the user has asked for a timestamp-based checkpoint.
+ if (endOffsets == null) {
+ endOffsets = consumer.endOffsets(topicPartitions);
+ }
Review Comment:
**Blocker, and this one is on me.** My nit asking for a lazy fetch
contradicted my own "do not swap those two lines" note in the same review. The
lazy form is the unsafe one.
`offsetsForTimes` now runs at :483, `endOffsets` at :502. A record appended
to a fallback partition between the two is dropped permanently: it did not
exist when `offsetsForTimes` ran, and the later tip is already past it. The
eager order is at-least-once. This is not a one-RPC window either -- both calls
retry through `HoodieRetryingKafkaConsumer` (`retry.max_count=4`,
`max_interval_ms=2000`).
The saving was not real anyway: this path only runs on the first batch after
an explicit `--checkpoint <ts>`, and `endOffsets` already fires at :344 and
:446.
```suggestion
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 {
// No record in this partition has a timestamp >= the requested one.
Fall back to the
// end offset (partition tip) rather than 0/earliest to avoid
re-consuming the whole
// partition. The tip is read BEFORE offsetsForTimes above on
purpose: a record landing
// between the two calls must be ingested, not skipped.
```
Keep it a separate call from `toOffsets` at :344 -- sharing one map there
would drop records for the opposite reason.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,17 +480,39 @@ private Option<String>
getOffsetsByTimestamp(KafkaConsumer consumer, List<Partit
.map(x -> new
TopicPartition(x.topic(), x.partition()))
.collect(Collectors.toMap(Function.identity(), x -> timestamp));
- Map<TopicPartition, Long> earliestOffsets =
consumer.beginningOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
+ // Fetch end offsets lazily: they are only needed when at least one
partition has no record
+ // at/after the requested timestamp. In the common case where every
partition resolves,
+ // this saves one round-trip to the broker on every sync.
+ Map<TopicPartition, Long> endOffsets = null;
+ // 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 (partition tip) rather than 0/earliest to avoid
re-consuming the whole
+ // partition when the user has asked for a timestamp-based checkpoint.
+ if (endOffsets == null) {
+ endOffsets = consumer.endOffsets(topicPartitions);
+ }
+ Long endOffset = endOffsets.get(map.getKey());
+ fallbackToEndOffsets.put(map.getKey(), endOffset);
+
sb.append(",").append(map.getKey().partition()).append(":").append(endOffset);
}
}
+ if (!fallbackToEndOffsets.isEmpty()) {
Review Comment:
**Blocker (coverage).** Nothing pins this WARN or the null-matrix behind it.
`getOffsetsByTimestamp` builds its consumer inside `getNextOffsetRanges`, so
the new tests can only reach it through a real broker.
This file already has the pattern, from HUDI-8955 (`6edf2094c4de`): the
sibling handler was made `@VisibleForTesting` and is driven with
`mock(KafkaConsumer.class)` over all-null / some-null / none-null via
`resolveEarliestOffsetsWithRetentionTestArgs`
(TestKafkaOffsetGen.java:650-676). `@VisibleForTesting` is already imported and
used at :524 and :647.
Please mark `getOffsetsByTimestamp` `@VisibleForTesting` and add the
equivalent `@ParameterizedTest`.
To be explicit so you do not chase it: the pre-0.10.0 case cannot be
reproduced on a broker here (`cp-kafka:7.7.1` is KRaft). A mock is the only way
to cover it.
##########
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, so no already-published "
+ + "data is re-consumed. "
Review Comment:
Doc nit: this states only the upside. The user-visible half is the other one
-- records already sitting in that partition that predate the checkpoint are
never ingested at all. On a bootstrap sync nothing was consumed yet, so "no
already-published data is re-consumed" does not describe what the operator will
see.
Also, `beginningOffsets` returns the log *start* offset, which is 0 only for
an untrimmed partition; "the beginning" is fine but "offset 0" elsewhere in the
description is imprecise.
```suggestion
+ "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. "
```
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -3334,30 +3338,133 @@ public void testJsonKafkaDFSSourceWithOffsets() throws
Exception {
@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;
+
+ // ---- Case 1: checkpoint captured BEFORE producing records ----
+ String topicName1 = "topic" + testNum;
+ topicName = topicName1;
Review Comment:
Nit: `topicName` is the inherited static field
(HoodieDeltaStreamerTestBase.java:142) and nothing here reads it --
`prepareJsonKafkaDFSFiles`, `prepareJsonKafkaDFSSource` and the new helper all
take an explicit topic parameter, and `@BeforeEach` resets it at :151.
Drop this assignment, the one at :3374, and the `topicName1` local.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -3334,30 +3338,133 @@ public void testJsonKafkaDFSSourceWithOffsets() throws
Exception {
@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;
+
+ // ---- Case 1: checkpoint captured BEFORE producing records ----
+ String topicName1 = "topic" + testNum;
+ topicName = topicName1;
+ long checkpointBeforeProduction = System.currentTimeMillis();
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, topicName1);
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName1);
+ String tableBasePath1 = basePath + "/test_json_kafka_table" + testNum;
HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ TestHelpers.makeConfig(tableBasePath1, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
true, 100000, false, null,
- null, "timestamp", String.valueOf(System.currentTimeMillis())),
jsc);
+ null, "timestamp", String.valueOf(checkpointBeforeProduction)),
jsc);
deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath, sqlContext);
+ assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath1, sqlContext);
+ deltaStreamer.shutdownGracefully();
- prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, false, topicName);
+ // ---- 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;
+ topicName = topicName2;
+ 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;
deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ 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)), jsc);
+ deltaStreamer.sync();
Review Comment:
Nit: this instance is synced then overwritten at :3392 without
`shutdownGracefully()`, so its write client is never closed. Same at :3421.
Inconsistent with :3366, :3401 and :3444 in this same method.
All five syncs here are one-shot, so `syncOnce(cfg)`
(HoodieDeltaStreamerTestBase.java:861) covers it -- it does new/sync/shutdown
and drops the local variable. Precedent in this file at :3309.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -3334,30 +3338,133 @@ public void testJsonKafkaDFSSourceWithOffsets() throws
Exception {
@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;
+
+ // ---- Case 1: checkpoint captured BEFORE producing records ----
+ String topicName1 = "topic" + testNum;
+ topicName = topicName1;
+ long checkpointBeforeProduction = System.currentTimeMillis();
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, topicName1);
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName1);
+ String tableBasePath1 = basePath + "/test_json_kafka_table" + testNum;
HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ TestHelpers.makeConfig(tableBasePath1, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
true, 100000, false, null,
- null, "timestamp", String.valueOf(System.currentTimeMillis())),
jsc);
+ null, "timestamp", String.valueOf(checkpointBeforeProduction)),
jsc);
deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath, sqlContext);
+ assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath1, sqlContext);
+ deltaStreamer.shutdownGracefully();
- prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, false, topicName);
+ // ---- 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;
+ topicName = topicName2;
+ 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;
deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ 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)), jsc);
+ deltaStreamer.sync();
+ 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);
+ deltaStreamer = new HoodieDeltaStreamer(
+ TestHelpers.makeConfig(tableBasePath2, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null, null,
+ "timestamp", null), jsc);
+ deltaStreamer.sync();
+ // 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);
+ deltaStreamer.shutdownGracefully();
Review Comment:
Optional, but strictly better than what is here. This second sync exists
only to prove the fallback checkpoint was persisted at the tip -- and the
discriminating assertion already fired at :3387 (`assertRecordCount(0, ...)`,
which would be 5 under the old code).
The repo asserts the stored kafka checkpoint directly in one line:
`TestHelpers.assertCommitMetadata` (HoodieDeltaStreamerTestBase.java:737), used
for exactly this shape at :988 -- `assertCommitMetadata(topicName +
",0:500,1:500", tableBasePath, 1)`.
Swapping both second syncs for that takes this method from 5 syncs to 2 and
asserts the exact checkpoint string instead of inferring it from a record count.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,17 +480,39 @@ private Option<String>
getOffsetsByTimestamp(KafkaConsumer consumer, List<Partit
.map(x -> new
TopicPartition(x.topic(), x.partition()))
.collect(Collectors.toMap(Function.identity(), x -> timestamp));
- Map<TopicPartition, Long> earliestOffsets =
consumer.beginningOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
+ // Fetch end offsets lazily: they are only needed when at least one
partition has no record
+ // at/after the requested timestamp. In the common case where every
partition resolves,
+ // this saves one round-trip to the broker on every sync.
+ Map<TopicPartition, Long> endOffsets = null;
+ // 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 (partition tip) rather than 0/earliest to avoid
re-consuming the whole
+ // partition when the user has asked for a timestamp-based checkpoint.
+ if (endOffsets == null) {
+ endOffsets = consumer.endOffsets(topicPartitions);
+ }
+ 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.keySet(), fallbackToEndOffsets);
Review Comment:
Nit, optional: this prints `fallbackToEndOffsets.keySet()` and then the
whole map, so the partition list appears twice -- `Map.toString()` already
contains the keys. The sibling handler at :556 logs it once.
Drop the `keySet()` argument and its `{}`, keeping only `Fallback offsets:
{}`.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java:
##########
@@ -3334,30 +3338,133 @@ public void testJsonKafkaDFSSourceWithOffsets() throws
Exception {
@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;
+
+ // ---- Case 1: checkpoint captured BEFORE producing records ----
+ String topicName1 = "topic" + testNum;
+ topicName = topicName1;
+ long checkpointBeforeProduction = System.currentTimeMillis();
+ prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, true, topicName1);
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName1);
+ String tableBasePath1 = basePath + "/test_json_kafka_table" + testNum;
HoodieDeltaStreamer deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ TestHelpers.makeConfig(tableBasePath1, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
true, 100000, false, null,
- null, "timestamp", String.valueOf(System.currentTimeMillis())),
jsc);
+ null, "timestamp", String.valueOf(checkpointBeforeProduction)),
jsc);
deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath, sqlContext);
+ assertRecordCount(JSON_KAFKA_NUM_RECORDS, tableBasePath1, sqlContext);
+ deltaStreamer.shutdownGracefully();
- prepareJsonKafkaDFSFiles(JSON_KAFKA_NUM_RECORDS, false, topicName);
+ // ---- 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;
+ topicName = topicName2;
+ 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;
deltaStreamer = new HoodieDeltaStreamer(
- TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ 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)), jsc);
+ deltaStreamer.sync();
+ 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);
+ deltaStreamer = new HoodieDeltaStreamer(
+ TestHelpers.makeConfig(tableBasePath2, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null, null,
+ "timestamp", null), jsc);
+ deltaStreamer.sync();
+ // 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);
+ deltaStreamer.shutdownGracefully();
+
+ // ---- Case 3: checkpoint captured between two partition-targeted batches
----
+ // Partition 0 receives its records BEFORE the checkpoint, partition 1
receives its records
+ // AFTER. Only partition 1 should be consumed on the first sync (partition
0 falls back to
+ // its end offset). A follow-up sync that adds fresh records to partition
0 should then
+ // ingest exactly those new records, proving partition 0's checkpoint was
stored at its tip
+ // rather than at offset 0.
+ String topicName3 = "topic_partial_" + testNum;
+ topicName = topicName3;
+ int numPartitions = 2;
+ int recordsPerPartition = JSON_KAFKA_NUM_RECORDS / numPartitions;
+ testUtils.createTopic(topicName3, numPartitions);
+ sendJsonMessagesToPartition(topicName3, 0, recordsPerPartition);
+ Thread.sleep(10);
+ long checkpointBetweenPartitions = System.currentTimeMillis();
+ Thread.sleep(10);
+ sendJsonMessagesToPartition(topicName3, 1, recordsPerPartition);
+ prepareJsonKafkaDFSSource(PROPS_FILENAME_TEST_JSON_KAFKA, "earliest",
topicName3);
+ String tableBasePath3 = basePath + "/test_json_kafka_table_partial_" +
testNum;
+ deltaStreamer = new HoodieDeltaStreamer(
+ TestHelpers.makeConfig(tableBasePath3, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
+ Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
+ true, 100000, false, null, null,
+ "timestamp", String.valueOf(checkpointBetweenPartitions)), jsc);
+ deltaStreamer.sync();
+ // Partition 0 predates the checkpoint => fallback to end offset (0
records). Partition 1
+ // was produced after the checkpoint => all recordsPerPartition records
ingested.
+ assertRecordCount(recordsPerPartition, tableBasePath3, sqlContext);
+
+ // Fresh records to partition 0 (produced after the earlier checkpoint)
should now be picked
+ // up. If we had incorrectly fallen back to offset 0 for partition 0
above, we'd have already
+ // consumed its earlier records and the assertion above would have failed
at 2 * recordsPerPartition.
+ sendJsonMessagesToPartition(topicName3, 0, recordsPerPartition);
+ deltaStreamer = new HoodieDeltaStreamer(
+ TestHelpers.makeConfig(tableBasePath3, WriteOperationType.UPSERT,
JsonKafkaSource.class.getName(),
Collections.emptyList(), PROPS_FILENAME_TEST_JSON_KAFKA, false,
true, 100000, false, null, null,
- "timestamp", String.valueOf(System.currentTimeMillis())), jsc);
+ "timestamp", null), jsc);
deltaStreamer.sync();
- assertRecordCount(JSON_KAFKA_NUM_RECORDS * 2, tableBasePath, sqlContext);
+ // Total: partition 1's original batch + partition 0's fresh batch.
Partition 0's original,
+ // pre-checkpoint batch stays skipped.
+ assertRecordCount(recordsPerPartition * 2L, tableBasePath3, sqlContext);
deltaStreamer.shutdownGracefully();
Review Comment:
Cleanliness, but a real one. Case 3 exercises the same production branch as
`testGetNextOffsetRangesFromTimestampCheckpointTypeWithPartialOffsets`, but
pays two full Spark syncs plus a third topic and table for it.
It is also the weaker of the two: `JSON_KAFKA_NUM_RECORDS = 5`
(HoodieDeltaStreamerTestBase.java:123), so `recordsPerPartition = 5 / 2 = 2`.
Case 3 discriminates on 2 vs 4 records where the unit test does 500 in one
direct call, and the integer division silently drops a record.
Please drop case 3 (:3403-3444), `sendJsonMessagesToPartition`, and the four
Kafka producer imports. Cases 1-2 plus the unit test already cover every branch.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java:
##########
@@ -186,6 +190,97 @@ public void
testGetNextOffsetRangesFromTimestampCheckpointType() {
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");
Review Comment:
Worth one more line. This asserts only `first.fromOffset()` and
`last.untilOffset()` per partition. With `numEvents = 500` and `minPartitions =
2`, partition 1 is deliberately split into sub-ranges -- and sub-range
accounting is the most-regressed code in this file: HUDI-7450 (#10768),
HUDI-7506 (#10869), HUDI-7511 (#10875), HUDI-7153 (#10205).
As written, overlapping or gapped middle sub-ranges would still pass. Add
`assertEquals(recordsPerPartition,
KafkaOffsetGen.CheckpointUtils.totalNewMessages(nextOffsetRanges));` and assert
p1's ranges are contiguous (`ranges[i].untilOffset() ==
ranges[i+1].fromOffset()`). Cheap, and it closes that exact bug class.
--
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]
