KPL-compatible aggregation
Packs multiple user records into a single Kinesis record using the same protobuf aggregation format as the Amazon KPL, so consumers using KCL deaggregation work unmodified.
Built on the official AWS SDK for Go V2, using the same aggregation format the KPL uses.
package main
import (
"context"
"log"
"net/http"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/kinesis"
producer "github.com/kinesis-producer-go/kinesis-producer"
)
func main() {
transport := http.DefaultTransport.(*http.Transport).Clone()
transport.MaxIdleConns = 20
transport.MaxIdleConnsPerHost = 20
httpClient := &http.Client{
Transport: transport,
}
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-west-2"), config.WithHTTPClient(httpClient))
if err != nil {
log.Fatalf("unable to load SDK config, %v", err)
}
client := kinesis.NewFromConfig(cfg)
pr := producer.New(&producer.Config{
StreamName: aws.String("test"),
BacklogCount: 2000,
Client: client,
})
pr.Start()
// Handle failures
go func() {
for r := range pr.NotifyFailures() {
// r contains `Data`, `PartitionKey` and `Error()`
log.Printf("failure record: %+v\n", r)
}
}()
for i := 0; i < 5000; i++ {
if err := pr.Put([]byte("foo")); err != nil {
log.Printf("error producing: %+v\n", err)
time.Sleep(1 * time.Second)
}
}
time.Sleep(1 * time.Minute)
pr.Stop()
}See Usage for the full producer.Config reference and logger adapters, CBOR + gzip transport for the optional optimized Kinesis client, or Aggregation Format for the on-wire record layout.