@@ -22,6 +22,7 @@ import (
2222 ydssource "github.com/transferia/transferia/pkg/providers/yds/source"
2323 "github.com/transferia/transferia/pkg/stats"
2424 "github.com/transferia/transferia/pkg/util"
25+ queues "github.com/transferia/transferia/pkg/util/queues"
2526 "github.com/transferia/transferia/pkg/util/queues/lbyds"
2627 "github.com/transferia/transferia/pkg/xtls"
2728 "go.ytsaurus.tech/library/go/core/log"
@@ -72,29 +73,15 @@ func (s *oneDCSource) Fetch() ([]abstract.ChangeItem, error) {
7273 case * persqueue.Data :
7374 var res []abstract.ChangeItem
7475 raw := make ([]abstract.ChangeItem , 0 )
75- parseWrapper := func (buffer []* persqueue.Data ) []abstract.ChangeItem {
76- for _ , item := range buffer {
77- for _ , b := range item .Batches () {
78- for _ , m := range b .Messages {
79- raw = append (raw , lbyds .MessageAsChangeItem (parsers.Message {
80- Offset : m .Offset ,
81- SeqNo : m .SeqNo ,
82- Key : m .SourceID ,
83- CreateTime : m .CreateTime ,
84- WriteTime : m .WriteTime ,
85- Value : m .Data ,
86- Headers : m .ExtraFields ,
87- }, parsers.MessageBatch {
88- Topic : b .Topic ,
89- Partition : b .Partition ,
90- Messages : nil , // not used here
91- }))
92- }
76+ parseWrapper := func (batches []parsers.MessageBatch ) []abstract.ChangeItem {
77+ for _ , messageBatch := range batches {
78+ for _ , message := range messageBatch .Messages {
79+ raw = append (raw , lbyds .MessageAsChangeItem (message , messageBatch , false ))
9380 }
9481 }
95- return lbyds .Parse (buffer , s .parser , s .metrics , s .logger , nil )
82+ return lbyds .Parse (batches , s .parser , s .metrics , s .logger , nil , false )
9683 }
97- parsed := parseWrapper ([] * persqueue. Data { v } )
84+ parsed := parseWrapper (lbyds . ConvertBatches ( v . Batches ()) )
9885 if len (raw ) > 3 {
9986 raw = raw [:3 ]
10087 }
@@ -262,9 +249,9 @@ func (s *oneDCSource) run(parseQ *parsequeue.WaitableParseQueue[batch]) error {
262249 s .logger .Debug ("got lb_offsets" , log .Any ("range" , lbyds .BuildMapPartitionToLbOffsetsRange (batches )))
263250
264251 s .lastRead = time .Now ()
265- messagesSize , messagesCount := BatchStatistics (batches )
266- s .metrics .Size .Add (int64 ( messagesSize ) )
267- s .metrics .Count .Add (int64 ( messagesCount ) )
252+ messagesSize , messagesCount := queues . BatchStatistics (batches )
253+ s .metrics .Size .Add (messagesSize )
254+ s .metrics .Count .Add (messagesCount )
268255 s .logger .Debugf ("Incoming data: %d messages of total size %d" , messagesCount , messagesSize )
269256
270257 splittedBatches := newBatches (s .maxBatchSize , v .Commit , batches )
@@ -539,6 +526,7 @@ func newPqv1Source(
539526 return ydssource .NewSourceWithOpts (transferID , ydsCfg , logger , registry ,
540527 ydssource .WithCreds (cfg .Credentials ),
541528 ydssource .WithReaderOpts (& readerOpts ),
529+ ydssource .WithUseFullTopicName (true ),
542530 ydssource .WithParser (parser ),
543531 )
544532}
0 commit comments