Refactor ndc history resender to handle multiple remote clusters (#2866)

common/xdc/nDCHistoryResender.go

@@ -30,6 +30,7 @@ import (
 	"context"
 	"time"
 
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/persistence/serialization"
 
 	commonpb "go.temporal.io/api/common/v1"
@@ -58,6 +59,7 @@ type (
 	NDCHistoryResender interface {
 		// SendSingleWorkflowHistory sends multiple run IDs's history events to remote
 		SendSingleWorkflowHistory(
+			remoteClusterName string,
 			namespaceID namespace.ID,
 			workflowID string,
 			runID string,
@@ -71,7 +73,7 @@ type (
 	// NDCHistoryResenderImpl is the implementation of NDCHistoryResender
 	NDCHistoryResenderImpl struct {
 		namespaceRegistry    namespace.Registry
-		adminClient          adminservice.AdminServiceClient
+		clientBean           client.Bean
 		historyReplicationFn nDCHistoryReplicationFn
 		serializer           serialization.Serializer
 		rereplicationTimeout dynamicconfig.DurationPropertyFnWithNamespaceIDFilter
@@ -91,7 +93,7 @@ const (
 // NewNDCHistoryResender create a new NDCHistoryResenderImpl
 func NewNDCHistoryResender(
 	namespaceRegistry namespace.Registry,
-	adminClient adminservice.AdminServiceClient,
+	clientBean client.Bean,
 	historyReplicationFn nDCHistoryReplicationFn,
 	serializer serialization.Serializer,
 	rereplicationTimeout dynamicconfig.DurationPropertyFnWithNamespaceIDFilter,
@@ -100,7 +102,7 @@ func NewNDCHistoryResender(
 
 	return &NDCHistoryResenderImpl{
 		namespaceRegistry:    namespaceRegistry,
-		adminClient:          adminClient,
+		clientBean:           clientBean,
 		historyReplicationFn: historyReplicationFn,
 		serializer:           serializer,
 		rereplicationTimeout: rereplicationTimeout,
@@ -110,6 +112,7 @@ func NewNDCHistoryResender(
 
 // SendSingleWorkflowHistory sends one run IDs's history events to remote
 func (n *NDCHistoryResenderImpl) SendSingleWorkflowHistory(
+	remoteClusterName string,
 	namespaceID namespace.ID,
 	workflowID string,
 	runID string,
@@ -131,13 +134,15 @@ func (n *NDCHistoryResenderImpl) SendSingleWorkflowHistory(
 
 	historyIterator := collection.NewPagingIterator(n.getPaginationFn(
 		ctx,
+		remoteClusterName,
 		namespaceID,
 		workflowID,
 		runID,
 		startEventID,
 		startEventVersion,
 		endEventID,
-		endEventVersion))
+		endEventVersion,
+	))
 
 	for historyIterator.HasNext() {
 		batch, err := historyIterator.Next()
@@ -172,6 +177,7 @@ func (n *NDCHistoryResenderImpl) SendSingleWorkflowHistory(
 
 func (n *NDCHistoryResenderImpl) getPaginationFn(
 	ctx context.Context,
+	remoteClusterName string,
 	namespaceID namespace.ID,
 	workflowID string,
 	runID string,
@@ -185,6 +191,7 @@ func (n *NDCHistoryResenderImpl) getPaginationFn(
 
 		response, err := n.getHistory(
 			ctx,
+			remoteClusterName,
 			namespaceID,
 			workflowID,
 			runID,
@@ -244,6 +251,7 @@ func (n *NDCHistoryResenderImpl) sendReplicationRawRequest(
 
 func (n *NDCHistoryResenderImpl) getHistory(
 	ctx context.Context,
+	remoteClusterName string,
 	namespaceID namespace.ID,
 	workflowID string,
 	runID string,
@@ -266,7 +274,13 @@ func (n *NDCHistoryResenderImpl) getHistory(
 
 	ctx, cancel := rpc.NewContextFromParentWithTimeoutAndHeaders(ctx, resendContextTimeout)
 	defer cancel()
-	response, err := n.adminClient.GetWorkflowExecutionRawHistoryV2(ctx, &adminservice.GetWorkflowExecutionRawHistoryV2Request{
+
+	adminClient, err := n.clientBean.GetRemoteAdminClient(remoteClusterName)
+	if err != nil {
+		return nil, err
+	}
+
+	response, err := adminClient.GetWorkflowExecutionRawHistoryV2(ctx, &adminservice.GetWorkflowExecutionRawHistoryV2Request{
 		Namespace: namespace.String(),
 		Execution: &commonpb.WorkflowExecution{
 			WorkflowId: workflowID,

common/xdc/nDCHistoryResender_test.go

@@ -29,8 +29,6 @@ import (
 	"testing"
 	"time"
 
-	"go.temporal.io/server/common/persistence/serialization"
-
 	"github.com/golang/mock/gomock"
 	"github.com/pborman/uuid"
 	"github.com/stretchr/testify/require"
@@ -45,10 +43,12 @@ import (
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/historyservicemock/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/primitives/timestamp"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 )
@@ -61,6 +61,7 @@ type (
 		controller          *gomock.Controller
 		mockClusterMetadata *cluster.MockMetadata
 		mockNamespaceCache  *namespace.MockRegistry
+		mockClientBean      *client.MockBean
 		mockAdminClient     *adminservicemock.MockAdminServiceClient
 		mockHistoryClient   *historyservicemock.MockHistoryServiceClient
 
@@ -91,10 +92,13 @@ func (s *nDCHistoryResenderSuite) SetupTest() {
 
 	s.controller = gomock.NewController(s.T())
 	s.mockClusterMetadata = cluster.NewMockMetadata(s.controller)
+	s.mockClientBean = client.NewMockBean(s.controller)
 	s.mockAdminClient = adminservicemock.NewMockAdminServiceClient(s.controller)
 	s.mockHistoryClient = historyservicemock.NewMockHistoryServiceClient(s.controller)
 	s.mockNamespaceCache = namespace.NewMockRegistry(s.controller)
 
+	s.mockClientBean.EXPECT().GetRemoteAdminClient(gomock.Any()).Return(s.mockAdminClient, nil).AnyTimes()
+
 	s.logger = log.NewTestLogger()
 	s.mockClusterMetadata.EXPECT().IsGlobalNamespaceEnabled().Return(true).AnyTimes()
 
@@ -118,7 +122,7 @@ func (s *nDCHistoryResenderSuite) SetupTest() {
 
 	s.rereplicator = NewNDCHistoryResender(
 		s.mockNamespaceCache,
-		s.mockAdminClient,
+		s.mockClientBean,
 		func(ctx context.Context, request *historyservice.ReplicateEventsV2Request) error {
 			_, err := s.mockHistoryClient.ReplicateEventsV2(ctx, request)
 			return err
@@ -219,6 +223,7 @@ func (s *nDCHistoryResenderSuite) TestSendSingleWorkflowHistory() {
 		}).Return(nil, nil).Times(2)
 
 	err := s.rereplicator.SendSingleWorkflowHistory(
+		cluster.TestCurrentClusterName,
 		s.namespaceID,
 		workflowID,
 		runID,
@@ -355,6 +360,7 @@ func (s *nDCHistoryResenderSuite) TestGetHistory() {
 
 	out, err := s.rereplicator.getHistory(
 		context.Background(),
+		cluster.TestCurrentClusterName,
 		s.namespaceID,
 		workflowID,
 		runID,

service/frontend/adminHandler.go

@@ -1419,13 +1419,9 @@ func (adh *AdminHandler) ResendReplicationTasks(
 	if request == nil {
 		return nil, adh.error(errRequestNotSet, scope)
 	}
-	remoteClient, err := adh.clientBean.GetRemoteAdminClient(request.GetRemoteCluster())
-	if err != nil {
-		return nil, err
-	}
 	resender := xdc.NewNDCHistoryResender(
 		adh.namespaceRegistry,
-		remoteClient,
+		adh.clientBean,
 		func(ctx context.Context, request *historyservice.ReplicateEventsV2Request) error {
 			_, err1 := adh.historyClient.ReplicateEventsV2(ctx, request)
 			return err1
@@ -1435,6 +1431,7 @@ func (adh *AdminHandler) ResendReplicationTasks(
 		adh.logger,
 	)
 	if err := resender.SendSingleWorkflowHistory(
+		request.GetRemoteCluster(),
 		namespace.ID(request.GetNamespaceId()),
 		request.GetWorkflowId(),
 		request.GetRunId(),

service/history/historyEngine.go

@@ -48,6 +48,7 @@ import (
 	"go.temporal.io/server/api/matchingservice/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	tokenspb "go.temporal.io/server/api/token/v1"
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
@@ -119,6 +120,7 @@ type (
 // NewEngineWithShardContext creates an instance of history engine
 func NewEngineWithShardContext(
 	shard shard.Context,
+	clientBean client.Bean,
 	matchingClient matchingservice.MatchingServiceClient,
 	sdkClientFactory sdk.ClientFactory,
 	eventNotifier events.Notifier,
@@ -213,7 +215,7 @@ func NewEngineWithShardContext(
 	)
 
 	historyEngImpl.workflowTaskHandler = newWorkflowTaskHandlerCallback(historyEngImpl)
-	historyEngImpl.replicationDLQHandler = replication.NewLazyDLQHandler(shard, workflowDeleteManager, historyCache)
+	historyEngImpl.replicationDLQHandler = replication.NewLazyDLQHandler(shard, workflowDeleteManager, historyCache, clientBean)
 
 	return historyEngImpl
 }

service/history/historyEngineFactory.go

@@ -27,6 +27,7 @@ package history
 import (
 	"go.uber.org/fx"
 
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/resource"
 	"go.temporal.io/server/common/sdk"
@@ -42,6 +43,7 @@ type (
 	HistoryEngineFactoryParams struct {
 		fx.In
 
+		ClientBean              client.Bean
 		MatchingClient          resource.MatchingClient
 		SdkClientFactory        sdk.ClientFactory
 		EventNotifier           events.Notifier
@@ -63,6 +65,7 @@ func (f *historyEngineFactory) CreateEngine(
 ) shard.Engine {
 	return NewEngineWithShardContext(
 		context,
+		f.ClientBean,
 		f.MatchingClient,
 		f.SdkClientFactory,
 		f.EventNotifier,

service/history/queueProcessorFactory.go

@@ -28,6 +28,7 @@ import (
 	"go.uber.org/fx"
 
 	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
@@ -81,6 +82,7 @@ type (
 
 		SchedulerParams
 
+		ClientBean       client.Bean
 		ArchivalClient   archiver.Client
 		SdkClientFactory sdk.ClientFactory
 		MatchingClient   resource.MatchingClient
@@ -92,6 +94,7 @@ type (
 
 		SchedulerParams
 
+		ClientBean     client.Bean
 		ArchivalClient archiver.Client
 		MatchingClient resource.MatchingClient
 	}
@@ -108,6 +111,7 @@ type (
 		fx.In
 
 		Config             *configs.Config
+		ClientBean         client.Bean
 		ArchivalClient     archiver.Client
 		EventSerializer    serialization.Serializer
 		TaskFetcherFactory replication.TaskFetcherFactory
@@ -177,9 +181,9 @@ func (f *transferQueueProcessorFactory) CreateProcessor(
 ) queues.Processor {
 	return newTransferQueueProcessor(
 		shard,
-		engine,
 		workflowCache,
 		f.scheduler,
+		f.ClientBean,
 		f.ArchivalClient,
 		f.SdkClientFactory,
 		f.MatchingClient,
@@ -228,9 +232,9 @@ func (f *timerQueueProcessorFactory) CreateProcessor(
 ) queues.Processor {
 	return newTimerQueueProcessor(
 		shard,
-		engine,
 		workflowCache,
 		f.scheduler,
+		f.ClientBean,
 		f.ArchivalClient,
 		f.MatchingClient,
 	)
@@ -305,5 +309,6 @@ func (f *replicationQueueProcessorFactory) CreateProcessor(
 		shard,
 		f.TaskFetcherFactory,
 		workflowCache,
+		f.ClientBean,
 	)
 }