luciano/vmware-go-kcl-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

milliseconds behind latest fix

Browse source
This commit is contained in:
ravikiran-s 2023-05-03 19:50:21 +05:30
parent ba3fde7e34
commit adb9ee1183
1 changed files with 9 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
clientlibrary/streams/dynamodb_adapter_client.go
View file

@ -10,6 +10,8 @@ import (
"github.com/aws/aws-sdk-go-v2/service/kinesis" "github.com/aws/aws-sdk-go-v2/service/kinesis"
ktypes "github.com/aws/aws-sdk-go-v2/service/kinesis/types" ktypes "github.com/aws/aws-sdk-go-v2/service/kinesis/types"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"math"
"time"
) )
type DynamodbStreamAdapterClient struct { type DynamodbStreamAdapterClient struct {
@ -299,6 +301,13 @@ func (d DynamodbStreamAdapterClient) convertGetRecordsOutput(dynamoOutput *dynam
kinesisOutput.Records = make([]ktypes.Record, 0) kinesisOutput.Records = make([]ktypes.Record, 0)
kinesisOutput.ResultMetadata = dynamoOutput.ResultMetadata kinesisOutput.ResultMetadata = dynamoOutput.ResultMetadata
kinesisOutput.NextShardIterator = dynamoOutput.NextShardIterator kinesisOutput.NextShardIterator = dynamoOutput.NextShardIterator
lastRecord := dynamoOutput.Records[len(dynamoOutput.Records)-1]
var lastRecordCreateTime *time.Time
if lastRecord.Dynamodb != nil && lastRecord.Dynamodb.ApproximateCreationDateTime != nil {
lastRecordCreateTime = lastRecord.Dynamodb.ApproximateCreationDateTime
}
millisBehindLatest := int64(math.Max(float64(time.Now().Sub(*lastRecordCreateTime).Milliseconds()), float64(0)))
kinesisOutput.MillisBehindLatest = &millisBehindLatest
} }
for _, record := range dynamoOutput.Records { for _, record := range dynamoOutput.Records {
reqBodyBytes := new(bytes.Buffer) reqBodyBytes := new(bytes.Buffer)