Ensure queue processor cluster ack level is below failover ack level (#…

…2794)

service/history/shard/context_impl.go

@@ -298,6 +298,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:
@@ -379,6 +382,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: