360
}
361
362
>
func (s *StreamSenderImpl) sendCatchUp(priority enumsspb.TaskPriority) (int64, error) {
stream_sender.go
363
>
readerID := shard.ReplicationReaderIDFromClusterShardID(
364
>
int64(s.clientShardKey.ClusterID),
365
>
s.clientShardKey.ShardID,
366
>
)
367
>
368
>
catchupEndExclusiveWatermark := s.shardContext.GetQueueExclusiveHighReadWatermark(tasks.CategoryReplication).TaskID
369
>
370
>
var catchupBeginInclusiveWatermark int64
371
>
queueState, ok := s.shardContext.GetQueueState(
372
>
tasks.CategoryReplication,
373
>
)
374
>
if !ok {
375
s.logger.Debug("StreamSender queueState not found")
376
catchupBeginInclusiveWatermark = catchupEndExclusiveWatermark
378
readerState, ok := queueState.ReaderStates[readerID]
379
if !ok {