fhan688 opened a new pull request, #20486:
URL: https://github.com/apache/hudi/pull/20486
### Purpose
`StreamingOffsetValidator` compares the Kafka offset difference between two
checkpoints with the total number of records written.
When a partition is added, removed, or has its offset reset
`KafkaCheckpointUtils` cannot determine the exact number of records consumed
from that partition. It currently skips such partitions and returns the offset
difference from only the comparable partitions.
The validator then compares this partial offset difference with the record
count written from all partitions. This can:
- Reject a valid commit when records from a new partition increase the total
write count.
- Accept a commit with missing records when records from a new partition
compensate for records lost from an existing partition.
This change makes incomplete checkpoint comparisons explicit and prevents
the validator from producing an incorrect pass or failure result.
### Brief change log
- Add `OffsetComparisonResult` to preserve both:
- The known offset difference from comparable partitions.
- Added, reset, and missing partitions that make the comparison incomplete.
- Add `KafkaCheckpointUtils.compareOffsets(...)` for callers that require an
exact offset comparison.
- Preserve the existing `calculateOffsetDifference(...)` signature and its
partial-difference behavior for compatibility.
- Update `StreamingOffsetValidator` to skip validation with a warning when
the checkpoint comparison is inconclusive.
- Continue performing the existing strict consistency validation when all
partitions are comparable.
- Update the offset comparison JavaDoc and remove the stale comment claiming
that reset partitions are calculated using their current offset.
### Tests
Added and updated unit tests covering:
- Exact offset comparisons with an unchanged partition set.
- Added Kafka partitions.
- Offset resets.
- Partitions missing from the current checkpoint.
- Added partitions in both Spark and Flink Kafka checkpoint formats.
- Compatibility of the existing `calculateOffsetDifference(...)` behavior.
- A valid commit that was previously rejected because records from a new
partition were compared against a partial offset difference.
- A compensating-count scenario that could previously be incorrectly
reported as validated.
- Existing exact-match and mismatch validation behavior.
Focused tests:
```shell
mvn -pl hudi-common,hudi-client/hudi-client-common \
-am \
-DskipITs \
-Dcheckstyle.skip=true \
-Dsurefire.failIfNoSpecifiedTests=false \
-Dtest=TestKafkaCheckpointUtils,TestStreamingOffsetValidator \
test
Results:
- TestKafkaCheckpointUtils: 23 tests passed.
- TestStreamingOffsetValidator: 27 tests passed.
- Maven reactor build succeeded.
- Checkstyle, RAT, and git diff --check passed.
Full mvn clean verify was not run.
```
### API and Format
This change adds the following public Java API:
- KafkaCheckpointUtils.OffsetComparisonResult
- KafkaCheckpointUtils.compareOffsets(...)
The API addition is backward compatible. The existing
`calculateOffsetDifference(...)` method retains its signature and return
semantics.
There are no changes to:
- Kafka checkpoint serialization formats.
- Hudi table or storage formats.
- Commit metadata formats.
- Configuration properties.
- RPC interfaces.
For incomplete comparisons, the validator now reports the result as
inconclusive and skips offset consistency validation instead of producing an
unreliable pass or failure.
### Documentation
No user-facing documentation changes are required because this is a
correctness fix and introduces no new configuration.
--
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]
