Close shard if get workflow execution failed on shardownershiplost error

cadence-workflow · Mar 16, 2024 · 7ffd197 · 7ffd197
1 parent 786d6e7
commit 7ffd197
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 2 deletions.
diff --git a/common/persistence/executionManager.go b/common/persistence/executionManager.go
@@ -73,6 +73,7 @@ func (m *executionManagerImpl) GetWorkflowExecution(
 	internalRequest := &InternalGetWorkflowExecutionRequest{
 		DomainID:  request.DomainID,
 		Execution: request.Execution,
+		RangeID: request.RangeID,
 	}
 	response, err := m.persistence.GetWorkflowExecution(ctx, internalRequest)
 	if err != nil {

diff --git a/service/history/engine/engineimpl/historyEngine.go b/service/history/engine/engineimpl/historyEngine.go
@@ -2835,7 +2835,12 @@ func (e *historyEngineImpl) ReplicateEventsV2(
 	replicateRequest *types.ReplicateEventsV2Request,
 ) error {
 
-	return e.nDCReplicator.ApplyEvents(ctx, replicateRequest)
+	err := e.nDCReplicator.ApplyEvents(ctx, replicateRequest)
+	if err != nil {
+		e.logger.Error("Failed to replicate events", tag.Error(err))
+		return err
+	}
+	return nil
 }
 
 func (e *historyEngineImpl) SyncShardStatus(

diff --git a/service/history/shard/context.go b/service/history/shard/context.go
@@ -590,7 +590,24 @@ func (s *contextImpl) GetWorkflowExecution(
 	}
 	currentRangeID := s.getRangeID()
 	request.RangeID = currentRangeID
-	return s.executionManager.GetWorkflowExecution(ctx, request)
+	response, err := s.executionManager.GetWorkflowExecution(ctx, request)
+	switch err.(type) {
+	case nil:
+		return response, nil
+	case *persistence.ShardOwnershipLostError:
+		{
+			// Shard is stolen, trigger shutdown of history engine
+			s.logger.Warn(
+				"Closing shard: GetWorkflowExecution failed due to stolen shard.",
+				tag.Error(err),
+				tag.ShardRangeID(currentRangeID),
+			)
+			s.closeShard()
+			return nil, err
+		}
+	default:
+		return nil, err
+	}
 }
 
 func (s *contextImpl) CreateWorkflowExecution(