You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2019/02/02 21:25:07 UTC
[GitHub] justinborromeo commented on a change in pull request #6917:
#6714-Implement-KCL-Deaggregation
justinborromeo commented on a change in pull request #6917: #6714-Implement-KCL-Deaggregation
URL: https://github.com/apache/incubator-druid/pull/6917#discussion_r253278978
##########
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:
The TeamCity build says that deaggregateKinesisRecords doesn't throw InvalidProtocolBufferException even though AggregatedRecordProtos#parseFrom throws the exception. From what I understand, the protobuf tests don't generate source files from .proto files at compile time...rather, someone compiled it once. That might be a valid option since we wouldn't expect the KPL aggregation schema to change anytime soon.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org