This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 6fc3ff2a5dae fix(kinesis): replace ASL-licensed KPL deaggregation with
a native decoder (#19707)
6fc3ff2a5dae is described below
commit 6fc3ff2a5dae70baf843bf406d1c8b99a94d0a44
Author: voonhous <[email protected]>
AuthorDate: Sat Aug 22 11:12:37 2026 +0800
fix(kinesis): replace ASL-licensed KPL deaggregation with a native decoder
(#19707)
* fix(kinesis): replace ASL-licensed KPL deaggregation with a native decoder
amazon-kinesis-deaggregator is under the Amazon Software License (ASF
Category X) and may not be a required dependency of an Apache project.
Decode the documented KPL aggregated-record format (magic prefix +
protobuf payload + trailing MD5) directly with protobuf-java instead,
and remove the now-dead AWS v1 SDK BOM pin and hudi-aws-bundle includes.
Differential-tested against the KCL implementation on a 17-case fixture
corpus. One deliberate deviation: a corrupt aggregate (valid digest,
out-of-range key index) passes through whole instead of KCL's silent
partial drop.
* fix(kinesis): address review round 1
- Fail the read with HoodieReadFromSourceException when a frame's KPL
digest verifies but the payload cannot be decoded, instead of passing
the raw frame through (which could land as an all-null row via
PERMISSIVE JSON parsing); digest-mismatch frames still pass through,
matching KCL, since an ordinary record can begin with the magic bytes.
- Check in two byte-frozen golden fixtures generated with AWS's
producer-side aggregation library (amazon-kinesis-aggregator) and
reference-decoded with the KCL deaggregator before freezing; the ASL
libraries were used at fixture-generation time only.
- Ban com.amazonaws:amazon-kinesis-client and amazon-kinesis-deaggregator
in the enforcer's bannedDependencies so the Category X artifacts cannot
return transitively (KCL 2.x lives at software.amazon.kinesis).
- Skip the SdkBytes defensive copy on the read path (asByteArrayUnsafe).
- Extract the shared KPL frame builders into KplTestUtils and annotate
the protobuf tag cases.
* fix(kinesis): address review round 2
- Reword the decode-failure message so enable.deaggregation=false reads
as a last-resort unblock: it ingests every aggregate frame raw and
loses the records inside them.
- Fail on a digest-verified aggregate that decodes to zero sub-records
instead of silently dropping the frame; the KPL never emits one, so
treat it like the other corruption paths.
---
hudi-utilities/pom.xml | 6 -
.../sources/helpers/KinesisDeaggregator.java | 184 +++++++++++--
.../utilities/sources/TestShardRecordIterator.java | 75 +++++-
.../utilities/sources/helpers/KplTestUtils.java | 101 +++++++
.../sources/helpers/TestKinesisDeaggregator.java | 290 +++++++++++++++++++++
packaging/hudi-aws-bundle/pom.xml | 3 -
pom.xml | 18 +-
7 files changed, 617 insertions(+), 60 deletions(-)
diff --git a/hudi-utilities/pom.xml b/hudi-utilities/pom.xml
index 35496f4051f2..ce64846a6534 100644
--- a/hudi-utilities/pom.xml
+++ b/hudi-utilities/pom.xml
@@ -578,12 +578,6 @@
<artifactId>sts</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
- <!-- KPL de-aggregation: extracts user records from Kinesis Producer
Library aggregated records -->
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>amazon-kinesis-deaggregator</artifactId>
- <version>${aws.kinesis.aggregator.version}</version>
- </dependency>
<!-- Hive - Test -->
<dependency>
<groupId>${hive.groupid}</groupId>
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java
index b0111f73f78e..d70e7d08d09f 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java
@@ -18,62 +18,196 @@
package org.apache.hudi.utilities.sources.helpers;
-import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord;
+import org.apache.hudi.utilities.config.KinesisSourceConfig;
+import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
+
+import com.google.protobuf.CodedInputStream;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.kinesis.model.Record;
-import java.nio.ByteBuffer;
+import java.io.IOException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
-import java.util.Date;
import java.util.List;
/**
* De-aggregates KPL (Kinesis Producer Library) aggregated records into
individual user records.
* Non-aggregated records are returned unchanged.
+ *
+ * <p>The aggregated record format (a 4-byte magic prefix, a protobuf payload
and a trailing MD5
+ * digest of that payload) is documented by the KPL and is decoded directly
here. This avoids a
+ * runtime dependency on the KCL de-aggregation library, which is published
under the Amazon
+ * Software License and therefore cannot be a required dependency of an Apache
project.
+ *
+ * <p>Semantics match the KCL deaggregator except for corrupt aggregates
(valid digest but an
+ * undecodable payload, an out-of-range key index or zero sub-records): KCL
keeps the sub-records
+ * before the bad one and silently drops the rest, while this implementation
fails the read, since a frame whose
+ * trailing digest verifies cannot be an ordinary user record and ingesting it
raw (or partially)
+ * would silently lose data. A frame that merely starts with the magic bytes
but whose digest does
+ * not verify is an ordinary user record and passes through unchanged, as with
KCL.
*/
public final class KinesisDeaggregator {
+ private static final byte[] MAGIC = new byte[] {(byte) 0xF3, (byte) 0x89,
(byte) 0x9A, (byte) 0xC2};
+ private static final int DIGEST_LENGTH = 16;
+
private KinesisDeaggregator() {
}
/**
* De-aggregate SDK v2 Kinesis records. Aggregated records (from KPL) are
split into user records.
* Non-aggregated records pass through unchanged.
+ *
+ * @throws HoodieReadFromSourceException if a record carries a valid KPL
aggregation digest but
+ * its payload cannot be decoded (corruption or an incompatible
aggregate format)
*/
public static List<Record> deaggregate(List<Record> records) {
if (records == null || records.isEmpty()) {
return new ArrayList<>();
}
- List<com.amazonaws.services.kinesis.model.Record> v1Records = new
ArrayList<>(records.size());
- for (Record r : records) {
- v1Records.add(toV1Record(r));
- }
- List<UserRecord> userRecords = UserRecord.deaggregate(v1Records);
- List<Record> result = new ArrayList<>(userRecords.size());
- for (UserRecord ur : userRecords) {
- result.add(toV2Record(ur));
+ List<Record> result = new ArrayList<>(records.size());
+ for (Record record : records) {
+ // Unsafe accessor skips SdkBytes' defensive copy; the array is only
read here, and
+ // sub-record payloads are copied out by readByteArray() before records
are built.
+ byte[] data = record.data() == null ? null :
record.data().asByteArrayUnsafe();
+ if (!isAggregated(data)) {
+ result.add(record);
+ continue;
+ }
+ int payloadLength = data.length - MAGIC.length - DIGEST_LENGTH;
+ try {
+ result.addAll(expand(record, data, MAGIC.length, payloadLength));
+ } catch (IOException e) {
+ throw new HoodieReadFromSourceException("Kinesis record with sequence
number " + record.sequenceNumber()
+ + " carries a valid KPL aggregation digest but could not be
decoded; this indicates corruption or an"
+ + " incompatible aggregate format, so the read is failed rather
than ingesting the raw frame."
+ + " As a last resort, " +
KinesisSourceConfig.KINESIS_ENABLE_DEAGGREGATION.key() + "=false unblocks"
+ + " the pipeline but ingests every aggregate frame raw, losing the
records inside them.", e);
+ }
}
return result;
}
- private static com.amazonaws.services.kinesis.model.Record toV1Record(Record
v2) {
- com.amazonaws.services.kinesis.model.Record v1 = new
com.amazonaws.services.kinesis.model.Record();
- v1.withData(ByteBuffer.wrap(v2.data().asByteArray()));
- v1.withPartitionKey(v2.partitionKey());
- v1.withSequenceNumber(v2.sequenceNumber());
- if (v2.approximateArrivalTimestamp() != null) {
-
v1.withApproximateArrivalTimestamp(Date.from(v2.approximateArrivalTimestamp()));
+ private static boolean isAggregated(byte[] data) {
+ // Strictly greater: a frame with an empty payload is not an aggregate,
matching KCL.
+ if (data == null || data.length <= MAGIC.length + DIGEST_LENGTH) {
+ return false;
+ }
+ for (int i = 0; i < MAGIC.length; i++) {
+ if (data[i] != MAGIC[i]) {
+ return false;
+ }
+ }
+ byte[] expectedDigest = new byte[DIGEST_LENGTH];
+ System.arraycopy(data, data.length - DIGEST_LENGTH, expectedDigest, 0,
DIGEST_LENGTH);
+ byte[] actualDigest = md5(data, MAGIC.length, data.length - MAGIC.length -
DIGEST_LENGTH);
+ return MessageDigest.isEqual(actualDigest, expectedDigest);
+ }
+
+ private static byte[] md5(byte[] data, int offset, int length) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("MD5");
+ digest.update(data, offset, length);
+ return digest.digest();
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("MD5 is not available in this JVM", e);
}
- return v1;
}
- private static Record toV2Record(UserRecord v1) {
+ /**
+ * Parses the {@code AggregatedRecord} message: repeated string
partition_key_table (field 1),
+ * repeated string explicit_hash_key_table (field 2) and repeated Record
records (field 3).
+ * Case labels are protobuf tags: (field number << 3) | wire type.
+ */
+ private static List<Record> expand(Record parent, byte[] data, int offset,
int length) throws IOException {
+ CodedInputStream input = CodedInputStream.newInstance(data, offset,
length);
+ List<String> partitionKeyTable = new ArrayList<>();
+ List<String> explicitHashKeyTable = new ArrayList<>();
+ List<byte[]> subMessages = new ArrayList<>();
+ while (!input.isAtEnd()) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 10: // field 1, length-delimited
+ partitionKeyTable.add(input.readStringRequireUtf8());
+ break;
+ case 18: // field 2, length-delimited
+ explicitHashKeyTable.add(input.readStringRequireUtf8());
+ break;
+ case 26: // field 3, length-delimited
+ subMessages.add(input.readByteArray());
+ break;
+ default:
+ input.skipField(tag);
+ break;
+ }
+ }
+ if (subMessages.isEmpty()) {
+ // Returning an empty list would silently drop the frame; the KPL never
emits an aggregate
+ // with zero records, so treat this like the other corruption paths.
+ throw new IOException("KPL aggregate contains no sub-records");
+ }
+ List<Record> expanded = new ArrayList<>(subMessages.size());
+ for (byte[] subMessage : subMessages) {
+ expanded.add(toRecord(parent, subMessage, partitionKeyTable,
explicitHashKeyTable));
+ }
+ return expanded;
+ }
+
+ /**
+ * Parses one nested {@code Record} message: required uint64
partition_key_index (field 1),
+ * optional uint64 explicit_hash_key_index (field 2), required bytes data
(field 3) and repeated
+ * Tag tags (field 4, skipped). Explicit hash keys and tags are validated
but not propagated.
+ * Case labels are protobuf tags: (field number << 3) | wire type.
+ */
+ private static Record toRecord(Record parent, byte[] subMessage,
List<String> partitionKeyTable,
+ List<String> explicitHashKeyTable) throws
IOException {
+ CodedInputStream input = CodedInputStream.newInstance(subMessage);
+ long partitionKeyIndex = 0;
+ boolean hasPartitionKeyIndex = false;
+ long explicitHashKeyIndex = 0;
+ boolean hasExplicitHashKeyIndex = false;
+ byte[] payload = null;
+ while (!input.isAtEnd()) {
+ int tag = input.readTag();
+ switch (tag) {
+ case 8: // field 1, varint
+ partitionKeyIndex = input.readUInt64();
+ hasPartitionKeyIndex = true;
+ break;
+ case 16: // field 2, varint
+ explicitHashKeyIndex = input.readUInt64();
+ hasExplicitHashKeyIndex = true;
+ break;
+ case 26: // field 3, length-delimited
+ payload = input.readByteArray();
+ break;
+ default:
+ input.skipField(tag);
+ break;
+ }
+ }
+ if (!hasPartitionKeyIndex) {
+ throw new IOException("KPL sub-record is missing the required partition
key index");
+ }
+ if (partitionKeyIndex < 0 || partitionKeyIndex >=
partitionKeyTable.size()) {
+ throw new IOException("KPL sub-record partition key index " +
partitionKeyIndex
+ + " is out of range for a table of " + partitionKeyTable.size() + "
keys");
+ }
+ if (hasExplicitHashKeyIndex
+ && (explicitHashKeyIndex < 0 || explicitHashKeyIndex >=
explicitHashKeyTable.size())) {
+ throw new IOException("KPL sub-record explicit hash key index " +
explicitHashKeyIndex
+ + " is out of range for a table of " + explicitHashKeyTable.size() +
" keys");
+ }
+ if (payload == null) {
+ throw new IOException("KPL sub-record is missing the required data
field");
+ }
Record.Builder builder = Record.builder()
- .data(SdkBytes.fromByteBuffer(v1.getData()))
- .partitionKey(v1.getPartitionKey())
- .sequenceNumber(v1.getSequenceNumber());
- if (v1.getApproximateArrivalTimestamp() != null) {
-
builder.approximateArrivalTimestamp(v1.getApproximateArrivalTimestamp().toInstant());
+ .data(SdkBytes.fromByteArray(payload))
+ .partitionKey(partitionKeyTable.get((int) partitionKeyIndex))
+ .sequenceNumber(parent.sequenceNumber());
+ if (parent.approximateArrivalTimestamp() != null) {
+
builder.approximateArrivalTimestamp(parent.approximateArrivalTimestamp());
}
return builder.build();
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestShardRecordIterator.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestShardRecordIterator.java
index d2cd528a83fd..415b92f7bf4a 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestShardRecordIterator.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestShardRecordIterator.java
@@ -19,6 +19,7 @@
package org.apache.hudi.utilities.sources;
import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
+import org.apache.hudi.utilities.sources.helpers.KplTestUtils;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -30,6 +31,7 @@ import
software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import
software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
import software.amazon.awssdk.services.kinesis.model.Record;
+import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
@@ -121,7 +123,7 @@ class TestShardRecordIterator {
@Test
void testMillisBehindLatestZeroStopsAfterCurrentPage() {
KinesisClient client = mock(KinesisClient.class);
- // Page 1: millisBehindLatest=0 → no second page should be fetched.
+ // Page 1: millisBehindLatest=0 -> no second page should be fetched.
when(client.getRecords(isA(GetRecordsRequest.class)))
.thenReturn(response(Arrays.asList(record("seq1"), record("seq2")),
NEXT_ITER, 0L));
@@ -199,7 +201,7 @@ class TestShardRecordIterator {
// -------------------------------------------------------------------------
/**
- * lastSequenceNumber must not advance until a page is fully consumed — it
commits only when
+ * lastSequenceNumber must not advance until a page is fully consumed -- it
commits only when
* hasNext() observes the current page is exhausted.
*/
@Test
@@ -213,14 +215,14 @@ class TestShardRecordIterator {
// Before any consumption no checkpoint yet.
assertFalse(it.getLastSequenceNumber().isPresent());
- it.next(); // seq1 — mid-page
+ it.next(); // seq1 -- mid-page
assertFalse(it.getLastSequenceNumber().isPresent());
it.next(); // seq2
- it.next(); // seq3 — page iterator is exhausted but commit hasn't fired yet
+ it.next(); // seq3 -- page iterator is exhausted but commit hasn't fired
yet
assertFalse(it.getLastSequenceNumber().isPresent());
- // hasNext() sees currentPage.hasNext() == false → commits
pendingPageLastSeq.
+ // hasNext() sees currentPage.hasNext() == false -> commits
pendingPageLastSeq.
assertFalse(it.hasNext());
assertEquals("seq3", it.getLastSequenceNumber().get());
}
@@ -240,7 +242,7 @@ class TestShardRecordIterator {
KinesisSource.ShardRecordIterator it = iterator(client, 100, 1000,
THROTTLE_TIMEOUT_LARGE);
it.next(); // seq1
- it.next(); // seq2 — exhausted page 1
+ it.next(); // seq2 -- exhausted page 1
// Page 1 not committed yet (hasNext() for seq3 commits it).
assertFalse(it.getLastSequenceNumber().isPresent());
@@ -277,10 +279,10 @@ class TestShardRecordIterator {
void testMultipleThrottlesHalveToFloorOfOne() {
KinesisClient client = mock(KinesisClient.class);
when(client.getRecords(isA(GetRecordsRequest.class)))
- .thenThrow(throttleEx()) // 8 → 4
- .thenThrow(throttleEx()) // 4 → 2
- .thenThrow(throttleEx()) // 2 → 1
- .thenThrow(throttleEx()) // 1 → 1 (floor)
+ .thenThrow(throttleEx()) // 8 -> 4
+ .thenThrow(throttleEx()) // 4 -> 2
+ .thenThrow(throttleEx()) // 2 -> 1
+ .thenThrow(throttleEx()) // 1 -> 1 (floor)
.thenReturn(response(Collections.singletonList(record("seq1")), null,
0L));
KinesisSource.ShardRecordIterator it = iterator(client, 8, 1000,
THROTTLE_TIMEOUT_LARGE);
@@ -315,7 +317,7 @@ class TestShardRecordIterator {
void testHalveAndHoldLimitForSubsequentPages() {
KinesisClient client = mock(KinesisClient.class);
when(client.getRecords(isA(GetRecordsRequest.class)))
- .thenThrow(throttleEx())
// call 1: throttled → halve 100→50
+ .thenThrow(throttleEx())
// call 1: throttled -> halve 100->50
.thenReturn(response(Collections.singletonList(record("seq1")),
NEXT_ITER, 5000L)) // call 2: success at 50
.thenReturn(response(Collections.singletonList(record("seq2")), null,
0L)); // call 3: next page, still at 50
@@ -329,7 +331,7 @@ class TestShardRecordIterator {
List<GetRecordsRequest> reqs = captor.getAllValues();
assertEquals(100, reqs.get(0).limit()); // initial attempt before throttle
assertEquals(50, reqs.get(1).limit()); // halved, succeeded
- assertEquals(50, reqs.get(2).limit()); // held — no recovery to 100
+ assertEquals(50, reqs.get(2).limit()); // held -- no recovery to 100
}
/**
@@ -361,4 +363,53 @@ class TestShardRecordIterator {
assertEquals(200, captor.getAllValues().get(0).limit());
assertEquals(100, captor.getAllValues().get(1).limit());
}
+
+ // -------------------------------------------------------------------------
+ // KPL de-aggregation
+ // -------------------------------------------------------------------------
+
+ /**
+ * Encodes a KPL aggregated record carrying one sub-record per payload
string, where payload i
+ * uses partition key index i. Frame building is shared with {@link
KplTestUtils}.
+ */
+ private static byte[] kplAggregate(List<String> partitionKeys, List<String>
payloads) throws Exception {
+ List<byte[]> subRecords = new ArrayList<>(payloads.size());
+ for (int i = 0; i < payloads.size(); i++) {
+ subRecords.add(KplTestUtils.encodeSubRecord(
+ i, null, payloads.get(i).getBytes(StandardCharsets.UTF_8), null));
+ }
+ return KplTestUtils.frame(KplTestUtils.encodeAggregatedRecord(
+ partitionKeys, Collections.emptyList(), subRecords));
+ }
+
+ @Test
+ void deaggregationEnabledFlattensKplAggregate() throws Exception {
+ KinesisClient client = mock(KinesisClient.class);
+ Record aggregated = Record.builder()
+ .data(SdkBytes.fromByteArray(kplAggregate(
+ Arrays.asList("pk-a", "pk-b"), Arrays.asList("{\"id\":1}",
"{\"id\":2}"))))
+ .sequenceNumber("seq-agg")
+ .partitionKey("parent-pk")
+ .approximateArrivalTimestamp(Instant.now())
+ .build();
+ when(client.getRecords(isA(GetRecordsRequest.class)))
+ .thenReturn(response(Collections.singletonList(aggregated), null, 0L));
+
+ KinesisSource.ShardRecordIterator it = new
KinesisSource.ShardRecordIterator(INITIAL_ITER, client,
+ SHARD_ID, 100, INTERVAL_MS, 1000, /* enableDeaggregation */ true,
+ RETRY_INITIAL_MS, RETRY_MAX_MS, THROTTLE_TIMEOUT_LARGE);
+
+ List<Record> collected = new ArrayList<>();
+ while (it.hasNext()) {
+ collected.add(it.next());
+ }
+
+ assertEquals(2, collected.size());
+ assertEquals("pk-a", collected.get(0).partitionKey());
+ assertEquals("{\"id\":1}", collected.get(0).data().asUtf8String());
+ assertEquals("pk-b", collected.get(1).partitionKey());
+ assertEquals("{\"id\":2}", collected.get(1).data().asUtf8String());
+ // Checkpoint must track the raw aggregated record's sequence number, not
a sub-record's.
+ assertEquals("seq-agg", it.getLastSequenceNumber().get());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KplTestUtils.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KplTestUtils.java
new file mode 100644
index 000000000000..de58a44e93a0
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KplTestUtils.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.sources.helpers;
+
+import com.google.protobuf.CodedOutputStream;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+
+/**
+ * Test-side encoder for the KPL aggregated record format (magic prefix,
AggregatedRecord protobuf
+ * payload, trailing MD5 digest), independent of the {@link
KinesisDeaggregator} decoder under test.
+ */
+public final class KplTestUtils {
+
+ public static final byte[] KPL_MAGIC = new byte[] {(byte) 0xF3, (byte) 0x89,
(byte) 0x9A, (byte) 0xC2};
+
+ private KplTestUtils() {
+ }
+
+ /**
+ * Encodes one nested {@code Record} message: partition key index (field 1),
optional explicit
+ * hash key index (field 2), payload (field 3) and an optional Tag (field 4).
+ */
+ public static byte[] encodeSubRecord(long pkIndex, Long ehkIndex, byte[]
data, String tagKeyOrNull)
+ throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ CodedOutputStream stream = CodedOutputStream.newInstance(out);
+ stream.writeUInt64(1, pkIndex);
+ if (ehkIndex != null) {
+ stream.writeUInt64(2, ehkIndex);
+ }
+ stream.writeByteArray(3, data);
+ if (tagKeyOrNull != null) {
+ ByteArrayOutputStream tagOut = new ByteArrayOutputStream();
+ CodedOutputStream tagStream = CodedOutputStream.newInstance(tagOut);
+ tagStream.writeString(1, tagKeyOrNull);
+ tagStream.flush();
+ stream.writeByteArray(4, tagOut.toByteArray());
+ }
+ stream.flush();
+ return out.toByteArray();
+ }
+
+ /**
+ * Encodes the {@code AggregatedRecord} message: partition key table (field
1), explicit hash key
+ * table (field 2) and the encoded sub-records (field 3).
+ */
+ public static byte[] encodeAggregatedRecord(List<String> pkTable,
List<String> ehkTable,
+ List<byte[]> subRecords) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ CodedOutputStream stream = CodedOutputStream.newInstance(out);
+ for (String pk : pkTable) {
+ stream.writeString(1, pk);
+ }
+ for (String ehk : ehkTable) {
+ stream.writeString(2, ehk);
+ }
+ for (byte[] subRecord : subRecords) {
+ stream.writeByteArray(3, subRecord);
+ }
+ stream.flush();
+ return out.toByteArray();
+ }
+
+ /** Wraps a payload in the KPL frame: magic prefix, payload, trailing MD5
digest of the payload. */
+ public static byte[] frame(byte[] payload) throws NoSuchAlgorithmException {
+ byte[] digest = MessageDigest.getInstance("MD5").digest(payload);
+ return ByteBuffer.allocate(KPL_MAGIC.length + payload.length +
digest.length)
+ .put(KPL_MAGIC).put(payload).put(digest).array();
+ }
+
+ /** Decodes a lower-case hex string, used for frozen byte-for-byte fixtures.
*/
+ public static byte[] hexToBytes(String hex) {
+ byte[] bytes = new byte[hex.length() / 2];
+ for (int i = 0; i < bytes.length; i++) {
+ bytes[i] = (byte) Integer.parseInt(hex.substring(i * 2, i * 2 + 2), 16);
+ }
+ return bytes;
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisDeaggregator.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisDeaggregator.java
new file mode 100644
index 000000000000..74361f56dae6
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisDeaggregator.java
@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.sources.helpers;
+
+import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.model.Record;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.hudi.utilities.sources.helpers.KplTestUtils.encodeAggregatedRecord;
+import static
org.apache.hudi.utilities.sources.helpers.KplTestUtils.encodeSubRecord;
+import static org.apache.hudi.utilities.sources.helpers.KplTestUtils.frame;
+import static
org.apache.hudi.utilities.sources.helpers.KplTestUtils.hexToBytes;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests for {@link KinesisDeaggregator}: KPL aggregate decoding
(including frozen fixtures
+ * from AWS's own producer-side aggregation library), the pass-through paths
taken by
+ * non-aggregated records, and the failure paths taken by corrupt aggregates.
+ */
+class TestKinesisDeaggregator {
+
+ private static final String PARENT_SEQ = "49590";
+ private static final Instant PARENT_ARRIVAL =
Instant.ofEpochMilli(1700000000000L);
+
+ // Golden fixtures produced by AWS's producer-side aggregation library
+ // (com.amazonaws:amazon-kinesis-aggregator:1.0.3, the Java implementation
of the KPL wire
+ // format) and reference-decoded with the KCL deaggregator
(amazon-kinesis-client:1.8.8) before
+ // being frozen here; both ASL-licensed libraries were used at
fixture-generation time only.
+ // The aggregator derives an explicit hash key for every record, so these
frames also populate
+ // the explicit hash key table and per-record indexes.
+
+ // Three sub-records: ("pk-a", {"id":1}), ("pk-b" with explicit hash key
+ // 170141183460469231731687303715884105727, {"id":2}), ("pk-a", {"id":3}).
+ private static final String KPL_PRODUCER_FRAME =
"f3899ac20a04706b2d610a04706b2d62122633373733343435363334393532"
+ +
"3538323733303632313239303034393331353131373531353612273137303134313138333436303436393233313733313638373330"
+ +
"333731353838343130353732371a0e080010001a087b226964223a317d1a0e080110011a087b226964223a327d1a0e080010001a08"
+ + "7b226964223a337d7d3f7b7fcad94b07edeb92fc05a4862d";
+
+ // Two sub-records exercising multi-byte UTF-8 (accented Latin and CJK) in
both the partition
+ // keys and the payloads; expected values are spelled with unicode escapes
in the test below.
+ private static final String KPL_PRODUCER_FRAME_UTF8 =
"f3899ac20a09706b2dc3bcc3b1c3ae0a08706b2d706c61696e122732"
+ +
"3333393233303635323234313430363231363337363735303732353237373832373039393731122635303738313230393835353930"
+ +
"383337363232303036343339313438313630383232393637391a18080010001a127b2263697479223a227ac3bc72696368227d1a1d"
+ +
"080110011a177b2263697479223a22746f6b796f20e69db1e4baac227dafed575c3266872e207cedb9cc6f3780";
+
+ private static Record kinesisRecord(byte[] data) {
+ return Record.builder()
+ .data(SdkBytes.fromByteArray(data))
+ .partitionKey("parent-pk")
+ .sequenceNumber(PARENT_SEQ)
+ .approximateArrivalTimestamp(PARENT_ARRIVAL)
+ .build();
+ }
+
+ private static byte[] utf8(String value) {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+
+ // -------------------------------------------------------------------------
+ // Tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ void deaggregatesAggregatedRecord() throws Exception {
+ List<byte[]> subRecords = Arrays.asList(
+ encodeSubRecord(0, null, utf8("{\"id\":1}"), null),
+ encodeSubRecord(1, null, utf8("{\"id\":2}"), null),
+ encodeSubRecord(0, null, utf8("{\"id\":3}"), null));
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Arrays.asList("pk-a", "pk-b"), Collections.emptyList(), subRecords)));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate));
+
+ assertEquals(3, result.size());
+ assertEquals("{\"id\":1}", result.get(0).data().asUtf8String());
+ assertEquals("{\"id\":2}", result.get(1).data().asUtf8String());
+ assertEquals("{\"id\":3}", result.get(2).data().asUtf8String());
+ assertEquals("pk-a", result.get(0).partitionKey());
+ assertEquals("pk-b", result.get(1).partitionKey());
+ assertEquals("pk-a", result.get(2).partitionKey());
+ for (Record record : result) {
+ assertEquals(PARENT_SEQ, record.sequenceNumber());
+ assertEquals(PARENT_ARRIVAL, record.approximateArrivalTimestamp());
+ }
+ }
+
+ @Test
+ void deaggregatesRealProducerLibraryFrame() {
+ Record aggregate = kinesisRecord(hexToBytes(KPL_PRODUCER_FRAME));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate));
+
+ assertEquals(3, result.size());
+ assertEquals("{\"id\":1}", result.get(0).data().asUtf8String());
+ assertEquals("{\"id\":2}", result.get(1).data().asUtf8String());
+ assertEquals("{\"id\":3}", result.get(2).data().asUtf8String());
+ assertEquals("pk-a", result.get(0).partitionKey());
+ assertEquals("pk-b", result.get(1).partitionKey());
+ assertEquals("pk-a", result.get(2).partitionKey());
+ for (Record record : result) {
+ assertEquals(PARENT_SEQ, record.sequenceNumber());
+ assertEquals(PARENT_ARRIVAL, record.approximateArrivalTimestamp());
+ }
+ }
+
+ @Test
+ void deaggregatesRealProducerLibraryFrameWithMultiByteUtf8() {
+ Record aggregate = kinesisRecord(hexToBytes(KPL_PRODUCER_FRAME_UTF8));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate));
+
+ assertEquals(2, result.size());
+ assertEquals("pk-\u00fc\u00f1\u00ee", result.get(0).partitionKey()); //
accented Latin partition key
+ assertEquals("{\"city\":\"z\u00fcrich\"}",
result.get(0).data().asUtf8String()); // accented Latin payload
+ assertEquals("pk-plain", result.get(1).partitionKey());
+ assertEquals("{\"city\":\"tokyo \u6771\u4eac\"}",
result.get(1).data().asUtf8String()); // CJK payload
+ }
+
+ @Test
+ void passesThroughNonAggregatedRecordUnchanged() {
+ Record plain = kinesisRecord(utf8("{\"id\":1,\"name\":\"plain\"}"));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(plain));
+
+ assertEquals(1, result.size());
+ assertSame(plain, result.get(0));
+ }
+
+ @Test
+ void passesThroughOnDigestMismatch() throws Exception {
+ // The magic prefix alone does not make an aggregate: an ordinary user
record could start with
+ // those four bytes, so a frame whose trailing digest does not verify
passes through, as in KCL.
+ byte[] framed = frame(encodeAggregatedRecord(
+ Collections.singletonList("pk-a"), Collections.emptyList(),
+ Collections.singletonList(encodeSubRecord(0, null, utf8("{\"id\":1}"),
null))));
+ framed[framed.length - 1] ^= 0x01;
+ Record corrupted = kinesisRecord(framed);
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(corrupted));
+
+ assertEquals(1, result.size());
+ assertSame(corrupted, result.get(0));
+ }
+
+ @Test
+ void failsOnCorruptPayloadWithValidDigest() throws Exception {
+ // Field 1, wire type 2, declared length 127 but only two bytes follow:
valid digest, bad protobuf.
+ Record corrupted = kinesisRecord(frame(new byte[] {0x0A, (byte) 0x7F,
0x01, 0x02}));
+
+ HoodieReadFromSourceException e =
assertThrows(HoodieReadFromSourceException.class,
+ () ->
KinesisDeaggregator.deaggregate(Collections.singletonList(corrupted)));
+ assertTrue(e.getMessage().contains(PARENT_SEQ));
+ }
+
+ @Test
+ void failsOnPartitionKeyIndexOutOfRange() throws Exception {
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Collections.singletonList("pk-a"), Collections.emptyList(),
+ Collections.singletonList(encodeSubRecord(5, null, utf8("{\"id\":1}"),
null)))));
+
+ assertThrows(HoodieReadFromSourceException.class,
+ () ->
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate)));
+ }
+
+ @Test
+ void failsOnExplicitHashKeyIndexOutOfRange() throws Exception {
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Collections.singletonList("pk-a"), Collections.emptyList(),
+ Collections.singletonList(encodeSubRecord(0, 3L, utf8("{\"id\":1}"),
null)))));
+
+ assertThrows(HoodieReadFromSourceException.class,
+ () ->
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate)));
+ }
+
+ @Test
+ void failsOnAggregateWithZeroSubRecords() throws Exception {
+ // Digest verifies and the payload parses, but there is no records field:
returning an empty
+ // list would make the frame vanish silently instead of failing like the
other corruption paths.
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Arrays.asList("pk-a", "pk-b"), Collections.emptyList(),
Collections.emptyList())));
+
+ assertThrows(HoodieReadFromSourceException.class,
+ () ->
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate)));
+ }
+
+ @Test
+ void decodesRecordWithExplicitHashKeysAndTags() throws Exception {
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Collections.singletonList("pk-a"),
+ Collections.singletonList("170141183460469231731687303715884105727"),
+ Collections.singletonList(encodeSubRecord(0, 0L, utf8("{\"id\":1}"),
"tag-key")))));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate));
+
+ assertEquals(1, result.size());
+ assertEquals("{\"id\":1}", result.get(0).data().asUtf8String());
+ assertEquals("pk-a", result.get(0).partitionKey());
+ }
+
+ @Test
+ void partiallyCorruptAggregateFailsWholeRead() throws Exception {
+ // Second sub-record is out of range: KCL would emit the first and
silently drop the rest,
+ // while this decoder fails the read so no part of a corrupt aggregate is
silently lost.
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Collections.singletonList("pk-a"), Collections.emptyList(),
+ Arrays.asList(
+ encodeSubRecord(0, null, utf8("{\"id\":1}"), null),
+ encodeSubRecord(5, null, utf8("{\"id\":2}"), null),
+ encodeSubRecord(0, null, utf8("{\"id\":3}"), null)))));
+
+ assertThrows(HoodieReadFromSourceException.class,
+ () ->
KinesisDeaggregator.deaggregate(Collections.singletonList(aggregate)));
+ }
+
+ @Test
+ void emptyPayloadWithValidDigestPassesThrough() throws Exception {
+ Record emptyPayload = kinesisRecord(frame(new byte[0]));
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Collections.singletonList(emptyPayload));
+
+ assertEquals(1, result.size());
+ assertSame(emptyPayload, result.get(0));
+ }
+
+ @Test
+ void tooShortDataPassesThrough() {
+ Record magicOnly = kinesisRecord(KplTestUtils.KPL_MAGIC);
+ Record tooShort = kinesisRecord(new byte[] {(byte) 0xF3, (byte) 0x89});
+
+ List<Record> result =
KinesisDeaggregator.deaggregate(Arrays.asList(magicOnly, tooShort));
+
+ assertEquals(2, result.size());
+ assertSame(magicOnly, result.get(0));
+ assertSame(tooShort, result.get(1));
+ }
+
+ @Test
+ void mixedBatchPreservesOrder() throws Exception {
+ Record first = kinesisRecord(utf8("{\"id\":\"first\"}"));
+ Record aggregate = kinesisRecord(frame(encodeAggregatedRecord(
+ Arrays.asList("pk-a", "pk-b"), Collections.emptyList(),
+ Arrays.asList(
+ encodeSubRecord(0, null, utf8("{\"id\":\"sub1\"}"), null),
+ encodeSubRecord(1, null, utf8("{\"id\":\"sub2\"}"), null)))));
+ Record last = kinesisRecord(utf8("{\"id\":\"last\"}"));
+
+ List<Record> result = KinesisDeaggregator.deaggregate(Arrays.asList(first,
aggregate, last));
+
+ assertEquals(4, result.size());
+ assertSame(first, result.get(0));
+ assertEquals("{\"id\":\"sub1\"}", result.get(1).data().asUtf8String());
+ assertEquals("pk-a", result.get(1).partitionKey());
+ assertEquals("{\"id\":\"sub2\"}", result.get(2).data().asUtf8String());
+ assertEquals("pk-b", result.get(2).partitionKey());
+ assertSame(last, result.get(3));
+ }
+
+ @Test
+ void emptyAndNullInputReturnEmptyList() {
+
assertTrue(KinesisDeaggregator.deaggregate(Collections.emptyList()).isEmpty());
+ assertTrue(KinesisDeaggregator.deaggregate(null).isEmpty());
+ }
+}
diff --git a/packaging/hudi-aws-bundle/pom.xml
b/packaging/hudi-aws-bundle/pom.xml
index 74bc4f71fba0..c3fd4b23643f 100644
--- a/packaging/hudi-aws-bundle/pom.xml
+++ b/packaging/hudi-aws-bundle/pom.xml
@@ -82,9 +82,6 @@
<include>org.apache.httpcomponents:httpcore</include>
<include>io.netty:*</include>
<include>software.amazon.awssdk:*</include>
- <!-- Kinesis KCL and deaggregation for
JsonKinesisSource -->
-
<include>com.amazonaws:amazon-kinesis-deaggregator</include>
-
<include>com.amazonaws:amazon-kinesis-client</include>
<include>io.dropwizard.metrics:metrics-core</include>
<include>com.beust:jcommander</include>
<include>commons-io:commons-io</include>
diff --git a/pom.xml b/pom.xml
index 52bc1149fd9b..5bb7060cc8ba 100644
--- a/pom.xml
+++ b/pom.xml
@@ -156,7 +156,6 @@
<prometheus.version>0.16.0</prometheus.version>
<aws.sdk.httpclient.version>4.5.13</aws.sdk.httpclient.version>
<aws.sdk.httpcore.version>4.4.13</aws.sdk.httpcore.version>
- <aws.kinesis.aggregator.version>1.0.3</aws.kinesis.aggregator.version>
<httpcore.version>4.4.16</httpcore.version>
<httpclient.version>4.5.14</httpclient.version>
<spark.version>${spark3.version}</spark.version>
@@ -245,7 +244,6 @@
<disruptor.version>3.4.2</disruptor.version>
<antlr.version>4.8</antlr.version>
<aws.sdk.version>2.29.52</aws.sdk.version>
- <aws.sdk.v1.version>1.12.797</aws.sdk.v1.version>
<proto.version>3.25.5</proto.version>
<protoc.version>3.25.5</protoc.version>
<dynamodb.lockclient.version>1.2.0</dynamodb.lockclient.version>
@@ -468,6 +466,10 @@
<exclude>org.apache.hbase:hbase-server:*:*:compile</exclude>
<!--To upgrade snappy because pre 1.1.8.2 does not work on
m1 mac-->
<exclude>org.xerial.snappy:snappy-java:*</exclude>
+ <!-- NOTE: Amazon Software License artifacts (ASF Category
X) must not come back, even
+ transitively; KCL 2.x moved to
software.amazon.kinesis and stays allowed -->
+ <exclude>com.amazonaws:amazon-kinesis-client</exclude>
+
<exclude>com.amazonaws:amazon-kinesis-deaggregator</exclude>
</excludes>
<includes>
<include>org.slf4j:slf4j-simple:*:*:test</include>
@@ -819,18 +821,6 @@
<dependencyManagement>
<dependencies>
- <!-- AWS v1 SDK BOM. Pins all com.amazonaws:aws-java-sdk-* artifacts to
a single
- version. Without this, transitive deps (notably
aws-lambda-java-events 1.1.0
- via amazon-kinesis-deaggregator) declare AWS SDK ranges like
[1.10.5,), which
- force Maven to walk every published patch version during
resolution. -->
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk-bom</artifactId>
- <version>${aws.sdk.v1.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
-
<!-- Scala -->
<dependency>
<groupId>org.scala-lang.modules</groupId>
