Fix handover callback (#3847)

Author: yycptt

service/history/historyEngine.go

@@ -301,7 +301,7 @@ func (e *historyEngineImpl) registerNamespaceStateChangeCallback() {
 
 	e.shard.GetNamespaceRegistry().RegisterStateChangeCallback(e, func(ns *namespace.Namespace, deletedFromDb bool) {
 		if e.shard.GetClusterMetadata().IsGlobalNamespaceEnabled() {
-			e.shard.UpdateHandoverNamespaces(ns, deletedFromDb)
+			e.shard.UpdateHandoverNamespace(ns, deletedFromDb)
 		}
 
 		if deletedFromDb {

service/history/shard/context.go

@@ -103,7 +103,7 @@ type (
 		// TODO: deprecate UpdateNamespaceNotificationVersion in v1.21 and remove
 		// NamespaceNotificationVersion from shardInfo proto blob
 		UpdateNamespaceNotificationVersion(namespaceNotificationVersion int64) error
-		UpdateHandoverNamespaces(ns *namespace.Namespace, deletedFromDb bool)
+		UpdateHandoverNamespace(ns *namespace.Namespace, deletedFromDb bool)
 
 		AppendHistoryEvents(ctx context.Context, request *persistence.AppendHistoryNodesRequest, namespaceID namespace.ID, execution commonpb.WorkflowExecution) (int, error)
 

service/history/shard/context_impl.go

@@ -578,11 +578,17 @@ func (s *ContextImpl) UpdateNamespaceNotificationVersion(namespaceNotificationVe
 	return nil
 }
 
-func (s *ContextImpl) UpdateHandoverNamespaces(ns *namespace.Namespace, deletedFromDb bool) {
+func (s *ContextImpl) UpdateHandoverNamespace(ns *namespace.Namespace, deletedFromDb bool) {
 	nsName := ns.Name()
+	// NOTE: replication state field won't be replicated and currently we only update a namespace
+	// to handover state from active cluster, so the second condition will always be true. Adding
+	// it here to be more safe in case above assumption no longer holds in the future.
+	isHandoverNamespace := ns.IsGlobalNamespace() &&
+		ns.ActiveInCluster(s.GetClusterMetadata().GetCurrentClusterName()) &&
+		ns.ReplicationState() == enums.REPLICATION_STATE_HANDOVER
 
 	s.wLock()
-	if deletedFromDb {
+	if deletedFromDb || !isHandoverNamespace {
 		delete(s.handoverNamespaces, ns.Name())
 		s.wUnlock()
 		return
@@ -595,23 +601,15 @@ func (s *ContextImpl) UpdateHandoverNamespaces(ns *namespace.Namespace, deletedF
 		maxReplicationTaskID = pendingMaxReplicationTaskID
 	}
 
-	// NOTE: replication state field won't be replicated and currently we only update a namespace
-	// to handover state from active cluster, so the second condition will always be true. Adding
-	// it here to be more safe in case above assumption no longer holds in the future.
-	if ns.IsGlobalNamespace() &&
-		ns.ActiveInCluster(s.GetClusterMetadata().GetCurrentClusterName()) &&
-		ns.ReplicationState() == enums.REPLICATION_STATE_HANDOVER {
-
-		if handover, ok := s.handoverNamespaces[nsName]; ok {
-			if handover.NotificationVersion < ns.NotificationVersion() {
-				handover.NotificationVersion = ns.NotificationVersion()
-				handover.MaxReplicationTaskID = maxReplicationTaskID
-			}
-		} else {
-			s.handoverNamespaces[nsName] = &namespaceHandOverInfo{
-				NotificationVersion:  ns.NotificationVersion(),
-				MaxReplicationTaskID: maxReplicationTaskID,
-			}
+	if handover, ok := s.handoverNamespaces[nsName]; ok {
+		if handover.NotificationVersion < ns.NotificationVersion() {
+			handover.NotificationVersion = ns.NotificationVersion()
+			handover.MaxReplicationTaskID = maxReplicationTaskID
+		}
+	} else {
+		s.handoverNamespaces[nsName] = &namespaceHandOverInfo{
+			NotificationVersion:  ns.NotificationVersion(),
+			MaxReplicationTaskID: maxReplicationTaskID,
 		}
 	}