voonhous commented on code in PR #19658:
URL: https://github.com/apache/hudi/pull/19658#discussion_r3811798557
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
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()));
+
sb.append(",").append(map.getKey().partition()).append(":").append(endOffsets.get(map.getKey()));
Review Comment:
**Blocker: this line breaks
`TestHoodieDeltaStreamer.testKafkaTimestampType`, and CI has already caught
it.**
Azure build 16360, job "UT Hudi Streamer & FT utilities":
```
[ERROR] TestHoodieDeltaStreamer.testKafkaTimestampType -- Time elapsed:
3.263 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <5> but was: <0>
at
HoodieDeltaStreamerTestBase.assertRecordCount(HoodieDeltaStreamerTestBase.java:553)
at
TestHoodieDeltaStreamer.testKafkaTimestampType(TestHoodieDeltaStreamer.java:3348)
```
It failed all 3 surefire retries, so it is deterministic, not flaky. The
GitHub Actions `test-utilities` lane is still queued on 14a4fdf, which is why
the PR does not look red yet.
Why it fails: the test publishes 5 records and only then evaluates
`String.valueOf(System.currentTimeMillis())` as the `--checkpoint`, so every
record predates T, `offsetsForTimes` returns null for both partitions, and this
line alone decides the checkpoint. Before: `0,0`, reads 5. After: end offsets,
`from == to`, `KafkaSource.toInputBatch` short-circuits at `totalNewMsgs <= 0`,
0 records committed.
This is not a stale assertion to quietly flip. `git show af837d2f1825` shows
[HUDI-1447] (#2438) added `getOffsetsByTimestamp`, the earliest fallback, and
this test in one commit, so that 5-record assertion is the written spec of the
contract you are changing.
**Action:** update `testKafkaTimestampType` in this PR to encode the new
contract, and say in the PR description that the old contract was changed
deliberately. Keep a case that proves ingestion still works (capture the
checkpoint timestamp *before* `prepareJsonKafkaDFSFiles` and keep
`assertRecordCount(JSON_KAFKA_NUM_RECORDS, ...)`), and add an explicit case
asserting `assertRecordCount(0, ...)` for a checkpoint captured after
production. Please do not just change `5` to `0` and `10` to `0`: that would
leave the test green even if the source were entirely broken.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
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()));
+
sb.append(",").append(map.getKey().partition()).append(":").append(endOffsets.get(map.getKey()));
Review Comment:
This is now a data-skipping path and it logs nothing. About 40 lines below,
the other `offsetsForTimes`-null handler in this same file does log it:
```java
List<TopicPartition> nullPartitions =
offsetAndTimestamp.entrySet().stream()
.filter(entry -> entry.getValue() == null)
.map(Map.Entry::getKey)
.collect(Collectors.toList());
if (!nullPartitions.isEmpty()) {
log.warn("OffsetAndTimestamp not available for partitions: {} since
{}", nullPartitions, retentionTs);
}
```
and the comment right under it documents the two causes: "message format
version before 0.10.0 or there is no data in the partition". The legacy-format
case is the one that matters here, since such a partition returns null for
*every* timestamp: it flips from "ingest everything" to "ingest nothing" while
the job commits a tip checkpoint, with nothing in the log to explain it. Rare
on modern brokers, but unguarded.
**Action:** collect the null partitions and emit one WARN naming them and
the end offsets chosen, mirroring the block above, and add a one-line comment
here stating that end (not beginning) is the correct start when no record has
ts >= T.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp =
consumer.offsetsForTimes(topicPartitionsTimestamp);
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()));
+
sb.append(",").append(map.getKey().partition()).append(":").append(endOffsets.get(map.getKey()));
Review Comment:
**Please add unit coverage for this branch.** Nothing in the repo currently
discriminates the line you changed.
`TestKafkaOffsetGen.testGetNextOffsetRangesFromTimestampCheckpointType`
(TestKafkaOffsetGen.java:175) uses `System.currentTimeMillis() - 100000`, a
*past* timestamp, so `offsetsForTimes` returns a non-null `OffsetAndTimestamp`
and the `if` branch is taken. It asserts `fromOffset == 0`, which holds under
both `beginningOffsets` and `endOffsets`, so it passes identically before and
after your change.
The mixed case (some partitions resolve, some return null) is the actual
user-visible bug and is covered nowhere at any level. Every prior bug in this
loop was a per-partition offset-string bug: f54b9bb8dd7c [HUDI-6191] #11686,
6edf2094c4de [HUDI-8955] #12762.
Two tests to add to `TestKafkaOffsetGen`, after line 187. Symbols and helper
signatures check out against master (`Helpers.jsonifyRecordsByPartitions` plus
the matching `sendMessages(String, Tuple2[])` overload,
`CheckpointUtils.totalNewMessages`, and the `testUtils` / `testTopicName` /
`metrics` fields), and no new imports are needed; please run them locally to
confirm.
```java
@Test
public void
testGetNextOffsetRangesFromTimestampCheckpointTypeWithNoOffsetsAfterTimestamp()
{
HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator();
testUtils.createTopic(testTopicName, 1);
testUtils.sendMessages(testTopicName,
Helpers.jsonifyRecords(dataGenerator.generateInserts("000", 1000)));
KafkaOffsetGen kafkaOffsetGen = new
KafkaOffsetGen(getConsumerConfigs("earliest", KAFKA_CHECKPOINT_TYPE_TIMESTAMP));
// No record has a timestamp >= this, so offsetsForTimes() returns null
for the partition and the
// fallback decides the start offset. 13 digits, as
isValidTimestampCheckpointType requires.
String futureTs = String.valueOf(System.currentTimeMillis() + 100000);
OffsetRange[] nextOffsetRanges = kafkaOffsetGen.getNextOffsetRanges(
Option.of(new StreamerCheckpointV2(futureTs)), 500, metrics);
assertEquals(1, nextOffsetRanges.length);
// fallback is endOffsets, not beginningOffsets: nothing new to read
assertEquals(1000, nextOffsetRanges[0].fromOffset());
assertEquals(1000, nextOffsetRanges[0].untilOffset());
}
@Test
public void
testGetNextOffsetRangesFromTimestampCheckpointTypeWithPartialOffsets() throws
InterruptedException {
HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator();
testUtils.createTopic(testTopicName, 2);
testUtils.sendMessages(testTopicName,
Helpers.jsonifyRecordsByPartitions(dataGenerator.generateInserts("000", 1000),
2));
Thread.sleep(50);
String ts = String.valueOf(System.currentTimeMillis());
Thread.sleep(50);
// all 100 records share one key, so only one of the two partitions has
data after ts
testUtils.sendMessages(testTopicName,
Helpers.jsonifyRecordsByPartitions(dataGenerator.generateInserts("001", 100),
1));
KafkaOffsetGen kafkaOffsetGen = new
KafkaOffsetGen(getConsumerConfigs("earliest", KAFKA_CHECKPOINT_TYPE_TIMESTAMP));
OffsetRange[] nextOffsetRanges = kafkaOffsetGen.getNextOffsetRanges(
Option.of(new StreamerCheckpointV2(ts)), Long.MAX_VALUE, metrics);
// the partition with no data after ts resumes at its end offset (500),
not at 0
assertEquals(100,
KafkaOffsetGen.CheckpointUtils.totalNewMessages(nextOffsetRanges));
}
```
Discrimination proof: pre-change the first yields `0/500` and the second
yields `600`, so both fail; post-change they yield `1000/1000` and `100`.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Review Comment:
**Design question worth settling before this merges.**
Hardcoding `endOffsets` makes the timestamp path the only branch in
`getNextOffsetRanges` that ignores `auto.offset.reset`, even though that is
precisely the "where do I start when I have no usable offset" knob, and it is
already resolved into `autoResetValue` (line 243) and switched on at line 323.
Routing the fallback through it instead:
```java
Map<TopicPartition, Long> fallbackOffsets;
switch (autoResetValue) {
case EARLIEST:
fallbackOffsets = consumer.beginningOffsets(topicPartitions);
break;
case LATEST:
fallbackOffsets = consumer.endOffsets(topicPartitions);
break;
case GROUP:
fallbackOffsets = getGroupOffsets(consumer, topicPartitions);
break;
default:
throw new HoodieNotSupportedException("Auto reset value must be one
of 'earliest' or 'latest' or 'group'");
}
```
gives you the fix by default, since `KAFKA_AUTO_OFFSET_RESET` defaults to
`LATEST` (KafkaSourceConfig.java:131), while leaving users who explicitly set
`earliest` on their current behavior. It also keeps `testKafkaTimestampType`
green as written, because that test sets `auto.offset.reset=earliest`
(TestHoodieDeltaStreamer.java:3340 -> :3218).
**Action:** either adopt this, or reply saying the unconditional skip-to-end
is intended regardless of `auto.offset.reset` and call the behavior change out
in the PR description. Both are defensible; it should be a stated decision
rather than an implicit one.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Review Comment:
Nit: the `getOffsetsByTimestamp` javadoc just above (lines 455-466) and the
`KAFKA_CHECKPOINT_TYPE` documentation string (KafkaSourceConfig.java:51-62,
"checkpoint should be provided as long value of desired timestamp") both say
nothing about what happens when a partition has no record at or after the
requested timestamp. That is exactly the ambiguity this PR resolves, so please
state the rule in one line in each. Feel free to split the config-doc half into
a separate change if you prefer.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -471,15 +471,15 @@ 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, Long> endOffsets =
consumer.endOffsets(topicPartitions);
Review Comment:
Nit, optional: `getNextOffsetRanges` issues this same
`consumer.endOffsets(topicPartitions)` call again a few frames later (line 345,
`toOffsets`). Not a regression, since the old code also made one unconditional
metadata RPC here, but it is now a duplicate of a call that happens anyway
rather than a distinct lookup. Could be computed lazily only when a null value
is actually present, or hoisted and shared with `toOffsets`. Feel free to
ignore.
--
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]
