[Claim] Add arrival time to Stream Record batch (#145)

v3io · Dec 20, 2023 · 89672a9 · 89672a9
1 parent 25cbc31
commit 89672a9
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 10 deletions.
diff --git a/pkg/dataplane/streamconsumergroup/claim.go b/pkg/dataplane/streamconsumergroup/claim.go
@@ -217,11 +217,13 @@ func (c *claim) fetchRecordBatch(location string) (string, error) {
 
 	for receivedRecordIndex, receivedRecord := range getRecordsOutput.Records {
 		record := v3io.StreamRecord{
-			ShardID:        &c.shardID,
-			Data:           receivedRecord.Data,
-			ClientInfo:     receivedRecord.ClientInfo,
-			PartitionKey:   receivedRecord.PartitionKey,
-			SequenceNumber: receivedRecord.SequenceNumber,
+			ShardID:         &c.shardID,
+			Data:            receivedRecord.Data,
+			ClientInfo:      receivedRecord.ClientInfo,
+			PartitionKey:    receivedRecord.PartitionKey,
+			SequenceNumber:  receivedRecord.SequenceNumber,
+			ArrivalTimeSec:  receivedRecord.ArrivalTimeSec,
+			ArrivalTimeNSec: receivedRecord.ArrivalTimeNSec,
 		}
 
 		records[receivedRecordIndex] = record

diff --git a/pkg/dataplane/types.go b/pkg/dataplane/types.go
@@ -326,11 +326,13 @@ type GetItemsOutput struct {
 //
 
 type StreamRecord struct {
-	ShardID        *int
-	Data           []byte
-	ClientInfo     []byte
-	PartitionKey   string
-	SequenceNumber uint64
+	ShardID         *int
+	Data            []byte
+	ClientInfo      []byte
+	PartitionKey    string
+	SequenceNumber  uint64
+	ArrivalTimeSec  int
+	ArrivalTimeNSec int
 }
 
 type SeekShardInputType int