wangxianghu commented on code in PR #19658:
URL: https://github.com/apache/hudi/pull/19658#discussion_r3831205040
##########
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:
the per-partition distribution from prepareJsonKafkaDFSFiles is
non-deterministic, so we can't hardcode the checkpoint string reliably
--
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]
