232
func (r *StreamReceiverImpl) ackMessage(
233
stream Stream,
235
>
highPriorityWaterMarkInfo := r.highPriorityTaskTracker.LowWatermark()
236
>
lowPriorityWaterMarkInfo := r.lowPriorityTaskTracker.LowWatermark()
237
>
size := r.highPriorityTaskTracker.Size() + r.lowPriorityTaskTracker.Size()
238
>
239
>
var highPriorityWatermark, lowPriorityWatermark *replicationspb.ReplicationState
240
>
inclusiveLowWaterMark := int64(-1)
241
>
var inclusiveLowWaterMarkTime time.Time
242
>
243
>
receiverMode := ReceiverMode(atomic.LoadInt32((*int32)(&r.receiverMode)))
244
>
switch receiverMode {
245
case ReceiverModeUnset:
246
return 0, nil