You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2026/05/20 06:21:42 UTC
(hudi) 29/45: fix: Follow-ups to JsonKinesisSource: numeric sequence comparison and call-site fixes (#18689)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-1.2.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 1558c870a908295a90a4c3d6b88162c9ee0fff86
Author: Lin Liu <14...@users.noreply.github.com>
AuthorDate: Fri May 15 14:23:19 2026 -0700
fix: Follow-ups to JsonKinesisSource: numeric sequence comparison and call-site fixes (#18689)
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---
.../hudi/utilities/sources/JsonKinesisSource.java | 4 ++--
.../hudi/utilities/sources/KinesisSource.java | 6 ++++--
.../sources/helpers/KinesisOffsetGen.java | 22 ++++++++++++++++++----
.../sources/helpers/KinesisReadConfig.java | 4 ++--
4 files changed, 26 insertions(+), 10 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
index 7ad947830097..3a297467c82e 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
@@ -143,12 +143,12 @@ public class JsonKinesisSource extends KinesisSource<JavaRDD<String>> {
KinesisSource.ShardRecordIterator recordIt = KinesisSource.readShardRecords(
client, readConfig.getStreamName(), range, readConfig.getStartingPosition(),
readConfig.getMaxRecordsPerRequest(), readConfig.getIntervalMilliSeconds(),
- readConfig.getMaxRecordsPerShard(), readConfig.isEnableDeaggregation(),
+ readConfig.getMaxRecordsPerShard(), readConfig.isDeaggregationEnabled(),
readConfig.getRetryInitialIntervalMs(), readConfig.getRetryMaxIntervalMs(),
readConfig.getThrottleTimeoutMs());
String shardId = range.getShardId();
- boolean addMetaFields = readConfig.isShouldAddMetaFields();
+ boolean addMetaFields = readConfig.isMetaFieldsEnabled();
List<String> jsonRecords = new ArrayList<>();
long numNull = 0;
java.time.Instant lastArrivalTimestamp = null;
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KinesisSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KinesisSource.java
index 919bac890581..a5ec35994d8e 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KinesisSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KinesisSource.java
@@ -237,7 +237,7 @@ public abstract class KinesisSource<T> extends Source<T> {
response = client.getRecords(
GetRecordsRequest.builder()
.shardIterator(shardIteratorStr)
- .limit(Math.min(currentMaxRecords, (int) (maxTotalRecords - totalConsumed)))
+ .limit(Math.min(currentMaxRecords, Math.toIntExact(maxTotalRecords - totalConsumed)))
.build());
lastSuccessTimeMs = System.currentTimeMillis();
break;
@@ -290,7 +290,9 @@ public abstract class KinesisSource<T> extends Source<T> {
// Process records first (done above), then decide whether to stop.
// millisBehindLatest can be 0 in LocalStack even when the response contained records.
- if (response.millisBehindLatest() == 0) {
+ // It is documented as nullable on AWS SDK responses, so guard against NPE on auto-unbox.
+ Long millisBehind = response.millisBehindLatest();
+ if (millisBehind != null && millisBehind == 0) {
fetchingDone = true;
}
if (shardIteratorStr == null) {
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
index b0f1779a50be..27d160c3de0f 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
@@ -42,6 +42,7 @@ import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceed
import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException;
import software.amazon.awssdk.services.kinesis.model.Shard;
+import java.math.BigInteger;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
@@ -162,6 +163,17 @@ public class KinesisOffsetGen {
Option.ofNullable(getEndSeqFromValue(value)));
}
+ /**
+ * Compares two Kinesis sequence numbers numerically.
+ * Kinesis sequence numbers are 128-bit integers represented as decimal strings whose lengths
+ * can vary, so lexicographic {@link String#compareTo} is not correct for numeric ordering.
+ *
+ * @return negative if a < b, zero if equal, positive if a > b
+ */
+ public static int compareSequenceNumbers(String a, String b) {
+ return new BigInteger(a).compareTo(new BigInteger(b));
+ }
+
/**
* Build checkpoint value without arrival time: "lastSeq" or "lastSeq|endSeq".
*/
@@ -262,7 +274,7 @@ public class KinesisOffsetGen {
return !useLatestWhenNoCheckpoint;
}
// CASE 3: Closed shard: lastSeq >= endSeq means fully consumed
- if (lastSeq.compareTo(endSeq) >= 0) {
+ if (CheckpointUtils.compareSequenceNumbers(lastSeq, endSeq) >= 0) {
return false;
}
// CASE 4: lastSeq < endSeq: may have unread records
@@ -459,9 +471,11 @@ public class KinesisOffsetGen {
Option<String> lastSeqOpt = seqs.getLeft();
Option<String> endSeqOpt = seqs.getRight();
// endSeq absent = was open shard; conservatively assume not fully consumed.
- // endSeq present: fully consumed iff lastSeq >= endSeq.
+ // endSeq present with non-empty lastSeq: fully consumed iff lastSeq >= endSeq.
+ // Empty lastSeq (e.g. "|endSeq" checkpoint) is treated as not consumed.
boolean fullyConsumed = endSeqOpt.isPresent()
- && lastSeqOpt.map(last -> last.compareTo(endSeqOpt.get()) >= 0).orElse(false);
+ && lastSeqOpt.map(last -> !last.isEmpty()
+ && CheckpointUtils.compareSequenceNumbers(last, endSeqOpt.get()) >= 0).orElse(false);
if (fullyConsumed) {
log.info("Expired shard {} was fully consumed (lastSeq >= endSeq); pruning from checkpoint", shardId);
} else {
@@ -495,7 +509,7 @@ public class KinesisOffsetGen {
continue;
}
// lastSeq < shardStartSeq: records between the checkpoint and the trim horizon were dropped.
- if (lastSeq.compareTo(shardStartSeq) < 0) {
+ if (CheckpointUtils.compareSequenceNumbers(lastSeq, shardStartSeq) < 0) {
reportDataLoss("Shard " + shardId + " checkpoint lastSeq=" + lastSeq
+ " is before the shard's earliest available sequence number " + shardStartSeq
+ ". Records may have been trimmed due to retention expiry (data loss).");
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
index e6bf12c5f8b1..7fe53f24d0f9 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
@@ -41,8 +41,8 @@ public class KinesisReadConfig implements Serializable {
private final String accessKey; // null if not set
private final String secretKey; // null if not set
private final KinesisSourceConfig.KinesisStartingPositionStrategy startingPosition;
- private final boolean shouldAddMetaFields;
- private final boolean enableDeaggregation;
+ private final boolean metaFieldsEnabled;
+ private final boolean deaggregationEnabled;
private final int maxRecordsPerRequest;
private final long intervalMilliSeconds;
private final long maxRecordsPerShard;