Comments (6)
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...
from vmware-go-kcl.
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
from vmware-go-kcl.
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.
from vmware-go-kcl.
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.
from vmware-go-kcl.
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.
from vmware-go-kcl.
Fair enough.
from vmware-go-kcl.
Related Issues (20)
- Lease stealing code exception HOT 3
- Option to assume role in AWS Kinesis client HOT 1
- Checkpointer interface breaking change in v1.3.0 HOT 1
- worker.go:339 Error in getRecords: unexpected EOF
- How to get the StreamName/ConsumerName inside ProcessRecords method HOT 1
- Unsupported protocol scheme
- Workload skewed across workers HOT 3
- (Still) Too many calls to DescribeStream HOT 3
- Error in getRecords leaves a dangling record processor. HOT 1
- Stuck in waitOnParentShard after resharding HOT 4
- Possible Data Race for input Checkpointer HOT 1
- Semantic versioning friendly version tags HOT 1
- Shard consumer sometimes fails to recover from error refreshing lease HOT 9
- Prometheus metrics should add appname as label and not in the metric name HOT 1
- Possibility to Work w/ DynamoDB Streams HOT 3
- Error in publishing cloudwatch metrics. Error: NoCredentialProviders: no valid providers in chain. HOT 1
- How to use record deaggregator? HOT 2
- Multiple consumers processing records from the same shard. HOT 10
- Tagged toolchain docker container vmware/go-kcl-toolchain:0.1.2 uses golang version 1.12.4
- AWS Go SDK V2 HOT 15
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from vmware-go-kcl.