Ensure queue processor cluster ack level is below failover ack level
@@ -340,6 +340,9 @@ func (s *ContextImpl) UpdateQueueAckLevel(
s.wLock()
defer s.wUnlock()
+ // the ack level is already the min ack level across all types of
+ // queue processors: active, passive, failover
+
// backward compatibility (for rollback)
switch category {
case tasks.CategoryTransfer:
@@ -421,6 +424,14 @@ func (s *ContextImpl) UpdateQueueClusterAckLevel(
s.wLock()
defer s.wUnlock()
+ if levels, ok := s.shardInfo.FailoverLevels[category]; ok {
+ for _, failoverLevel := range levels {
+ if ackLevel.CompareTo(failoverLevel.CurrentLevel) > 0 {
+ ackLevel = failoverLevel.CurrentLevel
+ }
+ }
+ }
+
// backward compatibility (for rollback)
switch category {
case tasks.CategoryTransfer:
