linliu-code commented on code in PR #19707:
URL: https://github.com/apache/hudi/pull/19707#discussion_r3833977908
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java:
##########
@@ -18,62 +18,190 @@
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 or an out-of-range key index): 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. Set "
+ + KinesisSourceConfig.KINESIS_ENABLE_DEAGGREGATION.key() + "=false
to pass raw records through.", e);
Review Comment:
The escape hatch in the message is worth a word of caution: setting
`enable.deaggregation=false` unblocks the pipeline, but it then ingests *every*
aggregate frame raw, which is the PERMISSIVE all-null outcome this throw exists
to prevent. Could the message frame it as a last-resort unblock rather than a
remedy?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisDeaggregator.java:
##########
@@ -18,62 +18,190 @@
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 or an out-of-range key index): 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. Set "
+ + KinesisSourceConfig.KINESIS_ENABLE_DEAGGREGATION.key() + "=false
to pass raw records through.", 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);
+ }
+ }
+
+ /**
+ * 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;
+ }
}
- return v1;
+ List<Record> expanded = new ArrayList<>(subMessages.size());
Review Comment:
A digest-verified frame whose payload decodes to zero sub-records returns an
empty list, so the frame disappears with no error -- the one remaining hole in
the "no silent loss on a digest-verified frame" principle. It matches KCL and
the KPL never emits such a frame, so this is latent rather than broken; just
flagging in case you'd rather it were symmetric with the other corruption paths.
--
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]
