1758
}
1759
1760
>
func (s *ContextImpl) getMaxReplicationTaskID() int64 {
context_impl.go
1761
>
return s.taskKeyManager.getExclusiveReaderHighWatermark(tasks.CategoryReplication).TaskID - 1
1762
>
}
1763
1764
>
func (s *ContextImpl) notifyReplicationQueueProcessor(taskID int64) {
context_impl.go
1765
>
// Replication ack level won't exceed the max taskID it received via task notification.
1766
>
// Since here we want it's ack level to advance to at least the input taskID, we need to
1767
>
// trigger an fake notification.
1768
>
1769
>
cancelledCtx, cancel := context.WithCancel(context.Background())
1770
>
cancel()
1771
>
1772
>
engine, err := s.engineFuture.Get(cancelledCtx)
1773
>
if err != nil {
1774
s.contextTaggedLogger.Warn("tried to notify replication queue processor when engine is not ready")
1775
return
1776
}
1777
1778
>
fakeReplicationTask := tasks.NewFakeTask(definition.WorkflowKey{}, tasks.CategoryReplication, tasks.MinimumKey.FireTime)
context_impl.go
1779
>
fakeReplicationTask.SetTaskID(taskID)
1780
>
1781
>
engine.NotifyNewTasks(map[tasks.Category][]tasks.Task{
1782
>
tasks.CategoryReplication: {fakeReplicationTask},
1783
>
})
1784
}
1785