Schedule a new workflow task after force flush buffer events (#4490)

* Schedule a new workflow task after force flush buffer events

service/history/ndc/branch_manager.go

@@ -190,6 +190,7 @@ func (r *BranchMgrImpl) flushBufferedEvents(
 	if err := targetWorkflow.FlushBufferedEvents(); err != nil {
 		return nil, 0, err
 	}
+
 	// the workflow must be updated as active, to send out replication tasks
 	if err := targetWorkflow.context.UpdateWorkflowExecutionAsActive(
 		ctx,

service/history/ndc/branch_manager_test.go

@@ -35,6 +35,7 @@ import (
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
 
+	enumsspb "go.temporal.io/server/api/enums/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/cluster"
@@ -254,6 +255,10 @@ func (s *branchMgrSuite) TestFlushBufferedEvents() {
 		"",
 		int64(0),
 	).Return(&historypb.HistoryEvent{}, nil)
+	s.mockMutableState.EXPECT().AddWorkflowTaskScheduledEvent(
+		false,
+		enumsspb.WORKFLOW_TASK_TYPE_NORMAL,
+	).Return(&workflow.WorkflowTaskInfo{}, nil)
 	s.mockMutableState.EXPECT().FlushBufferedEvents()
 	s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, lastWriteVersion).Return(cluster.TestCurrentClusterName).AnyTimes()
 	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()

service/history/ndc/workflow.go

@@ -220,7 +220,16 @@ func (r *WorkflowImpl) FlushBufferedEvents() error {
 		return serviceerror.NewInternal("Workflow encountered workflow with buffered events but last write not from current cluster")
 	}
 
-	return r.failWorkflowTask(lastWriteVersion)
+	if err = r.failWorkflowTask(lastWriteVersion); err != nil {
+		return err
+	}
+	if _, err := r.mutableState.AddWorkflowTaskScheduledEvent(
+		false,
+		enumsspb.WORKFLOW_TASK_TYPE_NORMAL,
+	); err != nil {
+		return err
+	}
+	return nil
 }
 
 func (r *WorkflowImpl) failWorkflowTask(