Support Kinesis aggregation format about vmware-go-kcl

[rayjlinden]

I get the issue - but not having the support for aggregated messages is a big hole. I was excited when I found this repo because I thought it might have what I was looking for. But now I'm sad. :(

Not at you though. proto being such a pain to use in go is the real issue...

[zollie]

You can deaggregate fairly easily with code like...

import (
   "bytes"
   "crypto/md5"
   "fmt"
   "github.com/aws/aws-sdk-go/service/kinesis"
   "github.com/awslabs/kinesis-aggregation/go/records"
   "github.com/gogo/protobuf/proto"
   //...
)

// magic number for KPL aggregate payload
var magicNum = []byte{0xF3, 0x89, 0x9A, 0xC2}

//...

// is kinesis record in KPL aggregation format
// see https://github.com/awslabs/amazon-kinesis-producer/blob/master/aggregation-format.md
func isAggregated(r *kinesis.Record) bool {
   if len(r.Data) < len(magicNum)+md5.Size {
   	return false
   }
   return bytes.Equal(r.Data[0:len(magicNum)], magicNum)
}

// deaggregate kinesis record
func deaggregate(rec *kinesis.Record) ([]*kinesis.Record, error) {
   data := rec.Data
   checkSum := fmt.Sprintf("%x", data[len(data)-md5.Size:])
   payload := data[len(magicNum) : len(data)-md5.Size]

   if fmt.Sprintf("%x", md5.Sum(payload)) != checkSum {
   	return nil, errors.New("kinesis aggregated record checksum did not match the value calculated from payload")
   }

   aggRecord := &records.AggregatedRecord{}
   err := proto.Unmarshal(payload, aggRecord)
   if err != nil {
   	return nil, err
   }

   recs := make([]*kinesis.Record, len(aggRecord.Records))

   for i, r := range aggRecord.Records {
   	kr := &kinesis.Record{
   		Data:                        r.Data,
   		PartitionKey:                &aggRecord.PartitionKeyTable[i],
   		SequenceNumber:              rec.SequenceNumber,
   		ApproximateArrivalTimestamp: rec.ApproximateArrivalTimestamp,
   		EncryptionType:              rec.EncryptionType,
   	}
   	recs[i] = kr
   }

   return recs, nil
}

This uses the protobuf structs from github.com/awslabs/kinesis-aggregation/go/records

[taoj-action]

The aggregation-format uses proto2 language (not proto3) but we use event-service (not yet open-sourced) instead of KPL for event publishing. event-service and all events are encoded using proto3. I haven't found a good way of making their work together yet.

[Jackyjjc]

In the JAVA version, it detects if the record is aggregated and if it is not, it uses the normal logic: https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/AggregatorUtil.java#L112

so it should work with both aggregated record and non-aggregated ones.

[taoj-action]

I knew that part. The issue is that our applications all use proto3 for encoding event. KPL uses proto2 encoding. I cannot make proto2 and proto3 working together in a single application.

[Jackyjjc]

Fair enough.