Delete global namespace (#2867)

temporalio · May 31, 2022 · 7af6cda · 7af6cda
1 parent 00394e9
commit 7af6cda
Show file tree

Hide file tree

Showing 6 changed files with 41 additions and 18 deletions.
diff --git a/common/namespace/transmissionTaskHandler.go b/common/namespace/transmissionTaskHandler.go
@@ -27,6 +27,7 @@ package namespace
 import (
 	"context"
 
+	enumspb "go.temporal.io/api/enums/v1"
 	namespacepb "go.temporal.io/api/namespace/v1"
 	replicationpb "go.temporal.io/api/replication/v1"
 
@@ -89,6 +90,10 @@ func (namespaceReplicator *namespaceReplicatorImpl) HandleTransmissionTask(
 	if len(replicationConfig.Clusters) <= 1 {
 		return nil
 	}
+	if info.State == enumspb.NAMESPACE_STATE_DELETED {
+		// Don't replicate deleted namespace changes.
+		return nil
+	}
 
 	taskType := enumsspb.REPLICATION_TASK_TYPE_NAMESPACE_TASK
 	task := &replicationspb.ReplicationTask_NamespaceTaskAttributes{

diff --git a/common/xdc/nDCHistoryResender.go b/common/xdc/nDCHistoryResender.go
@@ -30,19 +30,18 @@ import (
 	"context"
 	"time"
 
-	"go.temporal.io/server/client"
-	"go.temporal.io/server/common/persistence/serialization"
-
 	commonpb "go.temporal.io/api/common/v1"
 
 	"go.temporal.io/server/api/adminservice/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/collection"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/rpc"
 )
 

diff --git a/service/history/nDCHistoryReplicator.go b/service/history/nDCHistoryReplicator.go
@@ -381,7 +381,7 @@ func (r *nDCHistoryReplicatorImpl) applyEvents(
 			return r.applyNonStartEventsResetWorkflow(ctx, context, mutableState, task)
 
 		default:
-			// unable to get mutable state, return err so we can retry the task later
+			// unable to get mutable state, return err, so we can retry the task later
 			return err
 		}
 	}
@@ -392,7 +392,7 @@ func (r *nDCHistoryReplicatorImpl) applyStartEvents(
 	context workflow.Context,
 	releaseFn workflow.ReleaseCacheFunc,
 	task nDCReplicationTask,
-) (retError error) {
+) error {
 
 	namespaceEntry, err := r.namespaceRegistry.GetNamespaceByID(task.getNamespaceID())
 	if err != nil {

diff --git a/service/history/replication/ack_manager.go b/service/history/replication/ack_manager.go
@@ -590,7 +590,7 @@ func (p *ackMgrImpl) processReplication(
 			return nil, nil
 		}
 		return action(msBuilder)
-	case *serviceerror.NotFound:
+	case *serviceerror.NotFound, *serviceerror.NamespaceNotFound:
 		return nil, nil
 	default:
 		return nil, err

diff --git a/service/history/timerQueueStandbyTaskExecutor.go b/service/history/timerQueueStandbyTaskExecutor.go
@@ -527,7 +527,9 @@ func (t *timerQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 	if err != nil {
 		return err
 	}
-	if resendInfo.lastEventID != common.EmptyEventID && resendInfo.lastEventVersion != common.EmptyVersion {
+	if resendInfo.lastEventID == common.EmptyEventID || resendInfo.lastEventVersion == common.EmptyVersion {
+		err = serviceerror.NewInternal("timerQueueStandbyProcessor encountered empty historyResendInfo")
+	} else {
 		ns, err := t.registry.GetNamespaceByID(namespace.ID(taskInfo.GetNamespaceID()))
 		if err != nil {
 			return err
@@ -541,12 +543,18 @@ func (t *timerQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 			taskInfo.GetWorkflowID(),
 			taskInfo.GetRunID(),
 		); err != nil {
+			if _, isNotFound := err.(*serviceerror.NamespaceNotFound); isNotFound {
+				// Don't log NamespaceNotFound error because it is valid case, and return error to stop retry.
+				return err
+			}
+
 			t.logger.Error("Error refresh tasks from remote.",
 				tag.ShardID(t.shard.GetShardID()),
 				tag.WorkflowNamespaceID(taskInfo.GetNamespaceID()),
 				tag.WorkflowID(taskInfo.GetWorkflowID()),
 				tag.WorkflowRunID(taskInfo.GetRunID()),
-				tag.ClusterName(remoteClusterName))
+				tag.ClusterName(remoteClusterName),
+				tag.Error(err))
 		}
 
 		// NOTE: history resend may take long time and its timeout is currently
@@ -561,17 +569,20 @@ func (t *timerQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 			common.EmptyEventID,
 			common.EmptyVersion,
 		)
-	} else {
-		err = serviceerror.NewInternal("timerQueueStandbyProcessor encountered empty historyResendInfo")
 	}
 
 	if err != nil {
+		if _, isNotFound := err.(*serviceerror.NamespaceNotFound); isNotFound {
+			// Don't log NamespaceNotFound error because it is valid case, and return error to stop retry.
+			return err
+		}
 		t.logger.Error("Error re-replicating history from remote.",
 			tag.ShardID(t.shard.GetShardID()),
 			tag.WorkflowNamespaceID(taskInfo.GetNamespaceID()),
 			tag.WorkflowID(taskInfo.GetWorkflowID()),
 			tag.WorkflowRunID(taskInfo.GetRunID()),
-			tag.ClusterName(remoteClusterName))
+			tag.ClusterName(remoteClusterName),
+			tag.Error(err))
 	}
 
 	// return error so task processing logic will retry

diff --git a/service/history/transferQueueStandbyTaskExecutor.go b/service/history/transferQueueStandbyTaskExecutor.go
@@ -603,7 +603,9 @@ func (t *transferQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 	if err != nil {
 		return err
 	}
-	if resendInfo.lastEventID != common.EmptyEventID && resendInfo.lastEventVersion != common.EmptyVersion {
+	if resendInfo.lastEventID == common.EmptyEventID || resendInfo.lastEventVersion == common.EmptyVersion {
+		err = serviceerror.NewInternal("transferQueueStandbyProcessor encountered empty historyResendInfo")
+	} else {
 		ns, err := t.registry.GetNamespaceByID(namespace.ID(taskInfo.GetNamespaceID()))
 		if err != nil {
 			return err
@@ -617,12 +619,17 @@ func (t *transferQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 			taskInfo.GetWorkflowID(),
 			taskInfo.GetRunID(),
 		); err != nil {
+			if _, isNotFound := err.(*serviceerror.NamespaceNotFound); isNotFound {
+				// Don't log NamespaceNotFound error because it is valid case, and return error to stop retry.
+				return err
+			}
 			t.logger.Error("Error refresh tasks from remote.",
 				tag.ShardID(t.shard.GetShardID()),
 				tag.WorkflowNamespaceID(taskInfo.GetNamespaceID()),
 				tag.WorkflowID(taskInfo.GetWorkflowID()),
 				tag.WorkflowRunID(taskInfo.GetRunID()),
-				tag.ClusterName(remoteClusterName))
+				tag.ClusterName(remoteClusterName),
+				tag.Error(err))
 		}
 
 		// NOTE: history resend may take long time and its timeout is currently
@@ -637,19 +644,20 @@ func (t *transferQueueStandbyTaskExecutor) fetchHistoryFromRemote(
 			0,
 			0,
 		)
-	} else {
-		err = serviceerror.NewInternal(
-			"transferQueueStandbyProcessor encountered empty historyResendInfo",
-		)
 	}
 
 	if err != nil {
+		if _, isNotFound := err.(*serviceerror.NamespaceNotFound); isNotFound {
+			// Don't log NamespaceNotFound error because it is valid case, and return error to stop retry.
+			return err
+		}
 		t.logger.Error("Error re-replicating history from remote.",
 			tag.ShardID(t.shard.GetShardID()),
 			tag.WorkflowNamespaceID(taskInfo.GetNamespaceID()),
 			tag.WorkflowID(taskInfo.GetWorkflowID()),
 			tag.WorkflowRunID(taskInfo.GetRunID()),
-			tag.SourceCluster(remoteClusterName))
+			tag.SourceCluster(remoteClusterName),
+			tag.Error(err))
 	}
 
 	// return error so task processing logic will retry