fix TestCDCIterator_RestartOnPosition

cdc_iterator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"cloud.google.com/go/spanner"
 	database "cloud.google.com/go/spanner/admin/database/apiv1"
@@ -128,8 +129,11 @@ func (c *cdcIterator) startReader(ctx context.Context, streamID string) {
 		for _, changeRec := range result.ChangeRecords {
 			for _, dataChangeRec := range changeRec.DataChangeRecords {
 				for _, mod := range dataChangeRec.Mods {
+
+					// add a nanosecond so that the position doesn't include the record itself
+					start := dataChangeRec.CommitTimestamp.Add(time.Nanosecond)
 					position := common.CDCPosition{
-						Start:    dataChangeRec.CommitTimestamp,
+						Start:    start,
 						StreamID: streamID,
 					}
 

cdc_iterator_test.go

@@ -132,13 +132,12 @@ func TestCDCIterator_RestartOnPosition(t *testing.T) {
 
 	var latestPosition opencdc.Position
 
-	{ // read and ack 2 records
-		testutils.ReadAndAssertInsert(ctx, is, iterator, user1)
-		rec := testutils.ReadAndAssertInsert(ctx, is, iterator, user2)
-		teardown()
+	// read and ack 2 records
+	testutils.ReadAndAssertInsert(ctx, is, iterator, user1)
+	rec := testutils.ReadAndAssertInsert(ctx, is, iterator, user2)
+	teardown()
 
-		latestPosition = rec.Position
-	}
+	latestPosition = rec.Position
 
 	// then, try to read from the second record
 

test/testutils.go

@@ -301,12 +301,14 @@ func isDataEqual(is *is.I, a, b opencdc.Data) {
 }
 
 func assertMetadata(is *is.I, metadata opencdc.Metadata) {
+	is.Helper()
 	col, err := metadata.GetCollection()
 	is.NoErr(err)
 	is.Equal(col, "Singers")
 }
 
 func assertKey(is *is.I, rec opencdc.Record, singer Singer) {
+	is.Helper()
 	singerID := fmt.Sprint(singer.SingerID)
 	isDataEqual(is, rec.Key, opencdc.StructuredData{"SingerID": singerID})
 }