gianm commented on a change in pull request #6917:
#6714-Implement-KCL-Deaggregation
URL: https://github.com/apache/incubator-druid/pull/6917#discussion_r252522811
##########
File path:
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -747,6 +706,41 @@ private String
getSequenceNumberInternal(StreamPartition<String> partition, Stri
}
+ @VisibleForTesting
+ List<byte[]> deaggregateKinesisRecord(Record kinesisRecord) throws
InvalidProtocolBufferException
+ {
+ ByteBuffer kinesisRecordData = kinesisRecord.getData();
+ int recordSize = kinesisRecordData.position(0).remaining();
+ boolean validAggregateLength = (recordSize >
KPL_AGGREGATE_MAGIC_NUMBERS.length + KPL_AGGREGATE_CHECKSUM_LENGTH_IN_BYTES);
+ if (!validAggregateLength) {
+ ByteBuffer kinesisData = (ByteBuffer)
kinesisRecord.getData().position(0);
+ return Collections.singletonList(toByteArray(kinesisData));
+ } else {
+ byte[] magicNumbers = new byte[KPL_AGGREGATE_MAGIC_NUMBERS.length];
+ byte[] protobufMessage = new byte[recordSize -
KPL_AGGREGATE_CHECKSUM_LENGTH_IN_BYTES - KPL_AGGREGATE_MAGIC_NUMBERS.length];
+ byte[] checksum = new byte[KPL_AGGREGATE_CHECKSUM_LENGTH_IN_BYTES];
+
+ kinesisRecordData.get(magicNumbers, 0, magicNumbers.length);
+ kinesisRecordData.get(protobufMessage, 0, protobufMessage.length);
+ kinesisRecordData.get(checksum, 0, checksum.length);
+ byte[] messageHash = Hashing.md5().hashBytes(protobufMessage).asBytes();
+
+ //Check magic numbers and checksum value to see if the message is an
aggregate
+ if (Arrays.equals(magicNumbers, KPL_AGGREGATE_MAGIC_NUMBERS)
+ && Arrays.equals(messageHash, checksum)) {
+ List<byte[]> data = new ArrayList<>();
+ AggregatedRecordProtos.AggregatedRecord aggregatedRecord =
AggregatedRecordProtos.AggregatedRecord.parseFrom(protobufMessage);
Review comment:
What's the error look like? (The job looks like it was restarted recently.)
How do the protobuf tests make this work - I think they use generated sources
too?
If those avenues don't turn up anything interesting, maybe we can suppress
something.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
