add topic path

ydb-platform · Nov 14, 2024 · 32c0f06 · 32c0f06
1 parent 3c1861b
commit 32c0f06
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/cmd/aardappel/main.go b/cmd/aardappel/main.go
@@ -148,7 +148,7 @@ func doMain(ctx context.Context, config configInit.Config, srcDb *ydb.Driver, ds
 			xlog.Fatal(ctx, "Unable to init dst table")
 		}
 		xlog.Debug(ctx, "Start reading")
-		go topicReader.ReadTopic(ctx, uint32(i), reader, prc)
+		go topicReader.ReadTopic(ctx, config.Streams[i].SrcTopic, uint32(i), reader, prc)
 	}
 
 	lockExecutor := func(fn func(context.Context, table.Session, table.Transaction) error) error {

diff --git a/internal/reader/reader.go b/internal/reader/reader.go
@@ -20,7 +20,7 @@ type TopicData struct {
 	Resolved json.RawMessage `json:"resolved"`
 }
 
-func ReadTopic(ctx context.Context, readerId uint32, reader *topicreader.Reader, channel processor.Channel) {
+func ReadTopic(ctx context.Context, topicPath string, readerId uint32, reader *topicreader.Reader, channel processor.Channel) {
 	var mu sync.Mutex
 	lastHb := make(map[int64]uint64)
 	verifyStream := func(part int64, data types.TxData) {
@@ -34,8 +34,8 @@ func ReadTopic(ctx context.Context, readerId uint32, reader *topicreader.Reader,
 		hb := lastHb[part]
 		if hb != 0 && data.Step < hb {
 			errString := fmt.Sprintf("Unexpected step_id in stream, last hb step_id: %v,"+
-				"got tx {\"key\":%v,\"ts\":[%v,%v]}",
-				lastHb[part], serializeKey(data.KeyValues), data.Step, data.TxId)
+				"got tx {\"topic\":\"%v\",\"key\":%v,\"ts\":[%v,%v]}",
+				lastHb[part], topicPath, serializeKey(data.KeyValues), data.Step, data.TxId)
 			stopErr := channel.SaveReplicationState(ctx, processor.REPLICATION_FATAL_ERROR, errString)
 			if stopErr != nil {
 				xlog.Fatal(ctx, errString,