Repository: uber/cadence
Pull request #3503 (2020-09-09 06:41:43)

Remove kafka replication from history and cli
<!-- Describe what has changed in this PR -->
**What changed?**
Remove kafka replication from history and cli

<!-- Tell your future self why have you made these changes -->
**Why?**
The code is not needed

<!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? -->
**How did you test it?**


<!-- Assuming the worst case, what can be broken when deploying this change to production? -->
**Potential risks**
In the worst case, the replication will fail.


coveralls: 
[![Coverage Status](https://coveralls.io/builds/33346439/badge)](https://coveralls.io/builds/33346439)

Coverage increased (+0.2%) to 67.175% when pulling **7be23ccbe0eb46e3549761b11d536e54b24e53e0 on kafka_cleanup** into **f8001425f3ff1bec237eaac4e396e6cb5766287d on master**.


Patch:
diff --git a/cmd/server/cadence/server.go b/cmd/server/cadence/server.go
index eb8d0184a..40cff5e1c 100644
--- a/cmd/server/cadence/server.go
+++ b/cmd/server/cadence/server.go
@@ -143,7 +143,6 @@ func (s *server) startService() common.Daemon {
 		clusterMetadata.MasterClusterName,
 		clusterMetadata.CurrentClusterName,
 		clusterMetadata.ClusterInformation,
-		clusterMetadata.ReplicationConsumer,
 	)
 
 	if s.cfg.PublicClient.HostPort != "" {
diff --git a/common/cluster/metadata.go b/common/cluster/metadata.go
index df8bf8e52..a5925d2b6 100644
--- a/common/cluster/metadata.go
+++ b/common/cluster/metadata.go
@@ -51,8 +51,6 @@ type (
 		GetAllClusterInfo() map[string]config.ClusterInformation
 		// ClusterNameForFailoverVersion return the corresponding cluster name for a given failover version
 		ClusterNameForFailoverVersion(failoverVersion int64) string
-		// GetReplicationConsumerConfig returns the config for replication task consumer.
-		GetReplicationConsumerConfig() *config.ReplicationConsumerConfig
 	}
 
 	metadataImpl struct {
@@ -71,8 +69,6 @@ type (
 		clusterInfo map[string]config.ClusterInformation
 		// versionToClusterName contains all initial version -> corresponding cluster name
 		versionToClusterName map[int64]string
-		//replicationConsumer returns the config for replication task consumer.
-		replicationConsumer *config.ReplicationConsumerConfig
 	}
 )
 
@@ -84,7 +80,6 @@ func NewMetadata(
 	masterClusterName string,
 	currentClusterName string,
 	clusterInfo map[string]config.ClusterInformation,
-	replicationConsumer *config.ReplicationConsumerConfig,
 ) Metadata {
 
 	if len(clusterInfo) == 0 {
@@ -129,7 +124,6 @@ func NewMetadata(
 	return &metadataImpl{
 		logger:                   logger,
 		enableGlobalDomain:       enableGlobalDomain,
-		replicationConsumer:      replicationConsumer,
 		failoverVersionIncrement: failoverVersionIncrement,
 		masterClusterName:        masterClusterName,
 		currentClusterName:       currentClusterName,
@@ -203,11 +197,3 @@ func (metadata *metadataImpl) ClusterNameForFailoverVersion(failoverVersion int6
 	}
 	return clusterName
 }
-
-func (metadata *metadataImpl) GetReplicationConsumerConfig() *config.ReplicationConsumerConfig {
-	if metadata.replicationConsumer == nil {
-		return &config.ReplicationConsumerConfig{Type: config.ReplicationConsumerTypeKafka}
-	}
-
-	return metadata.replicationConsumer
-}
diff --git a/common/cluster/metadata_mock.go b/common/cluster/metadata_mock.go
index 4dcfaf945..2980f6c8a 100644
--- a/common/cluster/metadata_mock.go
+++ b/common/cluster/metadata_mock.go
@@ -169,17 +169,3 @@ func (mr *MockMetadataMockRecorder) ClusterNameForFailoverVersion(failoverVersio
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterNameForFailoverVersion", reflect.TypeOf((*MockMetadata)(nil).ClusterNameForFailoverVersion), failoverVersion)
 }
-
-// GetReplicationConsumerConfig mocks base method
-func (m *MockMetadata) GetReplicationConsumerConfig() *config.ReplicationConsumerConfig {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetReplicationConsumerConfig")
-	ret0, _ := ret[0].(*config.ReplicationConsumerConfig)
-	return ret0
-}
-
-// GetReplicationConsumerConfig indicates an expected call of GetReplicationConsumerConfig
-func (mr *MockMetadataMockRecorder) GetReplicationConsumerConfig() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetReplicationConsumerConfig", reflect.TypeOf((*MockMetadata)(nil).GetReplicationConsumerConfig))
-}
diff --git a/common/log/tag/values.go b/common/log/tag/values.go
index 94eeda04a..0c44807b2 100644
--- a/common/log/tag/values.go
+++ b/common/log/tag/values.go
@@ -112,6 +112,7 @@ var (
 	ComponentMatchingEngine           = component("matching-engine")
 	ComponentReplicator               = component("replicator")
 	ComponentReplicationTaskProcessor = component("replication-task-processor")
+	ComponentReplicationAckManager    = component("replication-ack-manager")
 	ComponentHistoryReplicator        = component("history-replicator")
 	ComponentHistoryResender          = component("history-resender")
 	ComponentIndexer                  = component("indexer")
diff --git a/common/messaging/kafkaProducer.go b/common/messaging/kafkaProducer.go
index 94c884aad..65a2f56ce 100644
--- a/common/messaging/kafkaProducer.go
+++ b/common/messaging/kafkaProducer.go
@@ -22,12 +22,10 @@ package messaging
 
 import (
 	"errors"
-	"fmt"
 
 	"github.com/Shopify/sarama"
 
 	"github.com/uber/cadence/.gen/go/indexer"
-	"github.com/uber/cadence/.gen/go/replicator"
 	"github.com/uber/cadence/common/codec"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
@@ -90,55 +88,8 @@ func (p *kafkaProducer) serializeThrift(input codec.ThriftObject) ([]byte, error
 	return payload, nil
 }
 
-func (p *kafkaProducer) getKeyForReplicationTask(task *replicator.ReplicationTask) sarama.Encoder {
-	if task == nil {
-		return nil
-	}
-
-	switch task.GetTaskType() {
-	case replicator.ReplicationTaskTypeHistory:
-		// Use workflowID as the partition key so all replication tasks for a workflow are dispatched to the same
-		// Kafka partition.  This will give us some ordering guarantee for workflow replication tasks at least at
-		// the messaging layer perspective
-		attributes := task.HistoryTaskAttributes
-		return sarama.StringEncoder(attributes.GetWorkflowId())
-	case replicator.ReplicationTaskTypeHistoryV2:
-		// Use workflowID as the partition key so all replication tasks for a workflow are dispatched to the same
-		// Kafka partition.  This will give us some ordering guarantee for workflow replication tasks at least at
-		// the messaging layer perspective
-		attributes := task.HistoryTaskV2Attributes
-		return sarama.StringEncoder(attributes.GetWorkflowId())
-	case replicator.ReplicationTaskTypeSyncActivity:
-		// Use workflowID as the partition key so all sync activity tasks for a workflow are dispatched to the same
-		// Kafka partition.  This will give us some ordering guarantee for workflow replication tasks atleast at
-		// the messaging layer perspective
-		attributes := task.SyncActivityTaskAttributes
-		return sarama.StringEncoder(attributes.GetWorkflowId())
-	case replicator.ReplicationTaskTypeHistoryMetadata,
-		replicator.ReplicationTaskTypeDomain,
-		replicator.ReplicationTaskTypeSyncShardStatus:
-		return nil
-	default:
-		panic(fmt.Sprintf("encounter unsupported replication task type: %v", task.GetTaskType()))
-	}
-
-	return nil
-}
-
 func (p *kafkaProducer) getProducerMessage(message interface{}) (*sarama.ProducerMessage, error) {
 	switch message := message.(type) {
-	case *replicator.ReplicationTask:
-		payload, err := p.serializeThrift(message)
-		if err != nil {
-			return nil, err
-		}
-		partitionKey := p.getKeyForReplicationTask(message)
-		msg := &sarama.ProducerMessage{
-			Topic: p.topic,
-			Key:   partitionKey,
-			Value: sarama.ByteEncoder(payload),
-		}
-		return msg, nil
 	case *indexer.Message:
 		payload, err := p.serializeThrift(message)
 		if err != nil {
diff --git a/common/messaging/noop_producer.go b/common/messaging/noop_producer.go
new file mode 100644
index 000000000..d509f2223
--- /dev/null
+++ b/common/messaging/noop_producer.go
@@ -0,0 +1,36 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package messaging
+
+type noopProducer struct{}
+
+// NewNoopProducer returns a no-op message producer
+func NewNoopProducer() Producer {
+	return &noopProducer{}
+}
+
+func (p noopProducer) Publish(
+	message interface{},
+) error {
+	return nil
+}
diff --git a/common/mocks/ClusterMetadata.go b/common/mocks/ClusterMetadata.go
index 76cab8c10..5197b9460 100644
--- a/common/mocks/ClusterMetadata.go
+++ b/common/mocks/ClusterMetadata.go
@@ -157,17 +157,3 @@ func (_m *ClusterMetadata) IsMasterCluster() bool {
 
 	return r0
 }
-
-// GetReplicationConsumerConfig provides a mock function with given fields:
-func (_m *ClusterMetadata) GetReplicationConsumerConfig() *config.ReplicationConsumerConfig {
-	ret := _m.Called()
-
-	var r0 *config.ReplicationConsumerConfig
-	if rf, ok := ret.Get(0).(func() *config.ReplicationConsumerConfig); ok {
-		r0 = rf()
-	} else {
-		r0 = ret.Get(0).(*config.ReplicationConsumerConfig)
-	}
-
-	return r0
-}
\ No newline at end of file
diff --git a/common/service/config/config.go b/common/service/config/config.go
index f6a6c8da1..ce869ad51 100644
--- a/common/service/config/config.go
+++ b/common/service/config/config.go
@@ -35,13 +35,6 @@ import (
 	"github.com/uber/cadence/common/service/dynamicconfig"
 )
 
-const (
-	// ReplicationConsumerTypeKafka means consuming replication tasks from kafka.
-	ReplicationConsumerTypeKafka = "kafka"
-	// ReplicationConsumerTypeRPC means pulling source DC for replication tasks.
-	ReplicationConsumerTypeRPC = "rpc"
-)
-
 type (
 	// Config contains the configuration for a set of cadence services
 	Config struct {
@@ -259,8 +252,6 @@ type (
 	// ClusterMetadata contains the all cluster which participated in cross DC
 	ClusterMetadata struct {
 		EnableGlobalDomain bool `yaml:"enableGlobalDomain"`
-		// ReplicationConsumerConfig determines how we consume replication tasks.
-		ReplicationConsumer *ReplicationConsumerConfig `yaml:"replicationConsumer"`
 		// FailoverVersionIncrement is the increment of each cluster version when failover happens
 		FailoverVersionIncrement int64 `yaml:"failoverVersionIncrement"`
 		// MasterClusterName is the master cluster name, only the master cluster can register / update domain
@@ -282,12 +273,6 @@ type (
 		RPCAddress string `yaml:"rpcAddress"`
 	}
 
-	// ReplicationConsumerConfig contains config for replication consumer
-	ReplicationConsumerConfig struct {
-		// Type determines how we consume replication tasks. It can be either kafka(default) or rpc.
-		Type string `yaml:"type"`
-	}
-
 	// ReplicationTaskProcessorConfig is the config for replication task processor.
 	ReplicationTaskProcessorConfig struct {
 		NoTaskInitialWaitIntervalSecs int     `yaml:"noTaskInitialWaitIntervalSecs"`
diff --git a/service/frontend/service.go b/service/frontend/service.go
index eb9172d9f..741baaddc 100644
--- a/service/frontend/service.go
+++ b/service/frontend/service.go
@@ -24,16 +24,12 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/stretchr/testify/mock"
-
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/client"
 	"github.com/uber/cadence/common/definition"
 	"github.com/uber/cadence/common/domain"
 	"github.com/uber/cadence/common/log"
-	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/messaging"
-	"github.com/uber/cadence/common/mocks"
 	"github.com/uber/cadence/common/persistence"
 	persistenceClient "github.com/uber/cadence/common/persistence/client"
 	espersistence "github.com/uber/cadence/common/persistence/elasticsearch"
@@ -228,21 +224,9 @@ func (s *Service) Start() {
 	var replicationMessageSink messaging.Producer
 	clusterMetadata := s.GetClusterMetadata()
 	if clusterMetadata.IsGlobalDomainEnabled() {
-		consumerConfig := clusterMetadata.GetReplicationConsumerConfig()
-		if consumerConfig != nil &&
-			consumerConfig.Type == config.ReplicationConsumerTypeRPC {
-			replicationMessageSink = s.GetDomainReplicationQueue()
-		} else {
-			var err error
-			replicationMessageSink, err = s.GetMessagingClient().NewProducerWithClusterName(
-				s.GetClusterMetadata().GetCurrentClusterName())
-			if err != nil {
-				logger.Fatal("Creating replicationMessageSink producer failed", tag.Error(err))
-			}
-		}
+		replicationMessageSink = s.GetDomainReplicationQueue()
 	} else {
-		replicationMessageSink = &mocks.KafkaProducer{}
-		replicationMessageSink.(*mocks.KafkaProducer).On("Publish", mock.Anything).Return(nil)
+		replicationMessageSink = messaging.NewNoopProducer()
 	}
 
 	wfHandler := NewWorkflowHandler(s, s.config, replicationMessageSink, client.NewVersionChecker())
diff --git a/service/history/conflictResolver.go b/service/history/conflictResolver.go
index 17b04e1d9..152a1649c 100644
--- a/service/history/conflictResolver.go
+++ b/service/history/conflictResolver.go
@@ -33,6 +33,10 @@ import (
 	"github.com/uber/cadence/service/history/shard"
 )
 
+var (
+	defaultHistoryPageSize = 1000
+)
+
 type (
 	conflictResolver interface {
 		reset(
diff --git a/service/history/engine/interface.go b/service/history/engine/interface.go
index 97b427f5c..20fe24275 100644
--- a/service/history/engine/interface.go
+++ b/service/history/engine/interface.go
@@ -81,6 +81,5 @@ type (
 		NotifyNewHistoryEvent(event *events.Notification)
 		NotifyNewTransferTasks(tasks []persistence.Task)
 		NotifyNewTimerTasks(tasks []persistence.Task)
-		NotifyNewReplicationTasks(tasks []persistence.Task)
 	}
 )
diff --git a/service/history/execution/context.go b/service/history/execution/context.go
index c9fb06236..dfb3a32ab 100644
--- a/service/history/execution/context.go
+++ b/service/history/execution/context.go
@@ -810,7 +810,6 @@ func (c *contextImpl) notifyTasks(
 	timerTasks []persistence.Task,
 ) {
 	c.shard.GetEngine().NotifyNewTransferTasks(transferTasks)
-	c.shard.GetEngine().NotifyNewReplicationTasks(replicationTasks)
 	c.shard.GetEngine().NotifyNewTimerTasks(timerTasks)
 }
 
diff --git a/service/history/handler.go b/service/history/handler.go
index f70df0049..aa614feea 100644
--- a/service/history/handler.go
+++ b/service/history/handler.go
@@ -130,7 +130,6 @@ func (h *Handler) Start() {
 	h.replicationTaskFetchers = replication.NewTaskFetchers(
 		h.GetLogger(),
 		h.config,
-		h.GetClusterMetadata().GetReplicationConsumerConfig(),
 		h.GetClusterMetadata(),
 		h.GetClientBean(),
 	)
diff --git a/service/history/historyEngine.go b/service/history/historyEngine.go
index 948ed5097..666e79fc5 100644
--- a/service/history/historyEngine.go
+++ b/service/history/historyEngine.go
@@ -53,7 +53,6 @@ import (
 	"github.com/uber/cadence/common/persistence"
 	checks "github.com/uber/cadence/common/reconciliation/common"
 	"github.com/uber/cadence/common/reconciliation/invariants"
-	sconfig "github.com/uber/cadence/common/service/config"
 	"github.com/uber/cadence/common/xdc"
 	"github.com/uber/cadence/service/history/config"
 	"github.com/uber/cadence/service/history/engine"
@@ -99,7 +98,6 @@ type (
 		replicator                *historyReplicator
 		nDCReplicator             ndc.HistoryReplicator
 		nDCActivityReplicator     ndc.ActivityReplicator
-		replicatorProcessor       ReplicatorQueueProcessor
 		historyEventNotifier      events.Notifier
 		tokenSerializer           common.TaskTokenSerializer
 		executionCache            *execution.Cache
@@ -112,6 +110,7 @@ type (
 		workflowResetter          reset.WorkflowResetter
 		queueTaskProcessor        task.Processor
 		replicationTaskProcessors []replication.TaskProcessor
+		replicationAckManager     replication.TaskAckManager
 		publicClient              workflowserviceclient.Interface
 		eventsReapplier           ndc.EventsReapplier
 		matchingClient            matching.Client
@@ -225,6 +224,10 @@ func NewEngineWithShardContext(
 		queueTaskProcessor:     queueTaskProcessor,
 		clientChecker:          client.NewVersionChecker(),
 		failoverMarkerNotifier: failoverMarkerNotifier,
+		replicationAckManager: replication.NewTaskAckManager(
+			shard,
+			executionCache,
+		),
 	}
 	historyEngImpl.decisionHandler = newDecisionHandler(historyEngImpl)
 	pRetry := checks.NewPersistenceRetryer(
@@ -279,14 +282,6 @@ func NewEngineWithShardContext(
 
 	// Only start the replicator processor if valid publisher is passed in
 	if publisher != nil {
-		historyEngImpl.replicatorProcessor = newReplicatorQueueProcessor(
-			shard,
-			historyEngImpl.executionCache,
-			publisher,
-			executionManager,
-			historyV2Manager,
-			logger,
-		)
 		historyEngImpl.replicator = newHistoryReplicator(
 			shard,
 			clock.NewRealTimeSource(),
@@ -397,11 +392,6 @@ func (e *historyEngineImpl) Start() {
 	// queue processor need to be started.
 	e.registerDomainFailoverCallback()
 
-	clusterMetadata := e.shard.GetClusterMetadata()
-	if e.replicatorProcessor != nil && clusterMetadata.GetReplicationConsumerConfig().Type != sconfig.ReplicationConsumerTypeRPC {
-		e.replicatorProcessor.Start()
-	}
-
 	for _, replicationTaskProcessor := range e.replicationTaskProcessors {
 		replicationTaskProcessor.Start()
 	}
@@ -417,9 +407,6 @@ func (e *historyEngineImpl) Stop() {
 
 	e.txProcessor.Stop()
 	e.timerProcessor.Stop()
-	if e.replicatorProcessor != nil {
-		e.replicatorProcessor.Stop()
-	}
 
 	for _, replicationTaskProcessor := range e.replicationTaskProcessors {
 		replicationTaskProcessor.Stop()
@@ -2800,15 +2787,6 @@ func (e *historyEngineImpl) NotifyNewTransferTasks(
 	}
 }
 
-func (e *historyEngineImpl) NotifyNewReplicationTasks(
-	tasks []persistence.Task,
-) {
-
-	if len(tasks) > 0 && e.replicatorProcessor != nil {
-		e.replicatorProcessor.notifyNewTask()
-	}
-}
-
 func (e *historyEngineImpl) NotifyNewTimerTasks(
 	tasks []persistence.Task,
 ) {
@@ -3112,7 +3090,7 @@ func (e *historyEngineImpl) GetReplicationMessages(
 	sw := e.metricsClient.StartTimer(scope, metrics.GetReplicationMessagesForShardLatency)
 	defer sw.Stop()
 
-	replicationMessages, err := e.replicatorProcessor.getTasks(
+	replicationMessages, err := e.replicationAckManager.GetTasks(
 		ctx,
 		pollingCluster,
 		lastReadMessageID,
@@ -3141,7 +3119,7 @@ func (e *historyEngineImpl) GetDLQReplicationMessages(
 
 	tasks := make([]*r.ReplicationTask, 0, len(taskInfos))
 	for _, taskInfo := range taskInfos {
-		task, err := e.replicatorProcessor.getTask(ctx, taskInfo)
+		task, err := e.replicationAckManager.GetTask(ctx, taskInfo)
 		if err != nil {
 			e.logger.Error("Failed to fetch DLQ replication messages.", tag.Error(err))
 			return nil, err
diff --git a/service/history/replication/task_ack_manager.go b/service/history/replication/task_ack_manager.go
new file mode 100644
index 000000000..f729c5b4d
--- /dev/null
+++ b/service/history/replication/task_ack_manager.go
@@ -0,0 +1,716 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+//go:generate mockgen -copyright_file ../../../LICENSE -package $GOPACKAGE -source $GOFILE -destination task_ack_manager_mock.go
+
+package replication
+
+import (
+	ctx "context"
+	"errors"
+	"strconv"
+	"time"
+
+	"github.com/uber/cadence/.gen/go/replicator"
+	"github.com/uber/cadence/.gen/go/shared"
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/backoff"
+	"github.com/uber/cadence/common/collection"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/quotas"
+	"github.com/uber/cadence/common/service/dynamicconfig"
+	exec "github.com/uber/cadence/service/history/execution"
+	"github.com/uber/cadence/service/history/shard"
+	"github.com/uber/cadence/service/history/task"
+)
+
+var (
+	errUnknownQueueTask       = errors.New("unknown task type")
+	errUnknownReplicationTask = errors.New("unknown replication task")
+	defaultHistoryPageSize    = 1000
+)
+
+type (
+	// TaskAckManager is the ack manager for replication tasks
+	TaskAckManager interface {
+		GetTask(
+			ctx ctx.Context,
+			taskInfo *replicator.ReplicationTaskInfo,
+		) (*replicator.ReplicationTask, error)
+
+		GetTasks(
+			ctx ctx.Context,
+			pollingCluster string,
+			lastReadTaskID int64,
+		) (*replicator.ReplicationMessages, error)
+	}
+
+	taskAckManagerImpl struct {
+		shard            shard.Context
+		executionCache   *exec.Cache
+		executionManager persistence.ExecutionManager
+		historyManager   persistence.HistoryManager
+		rateLimiter      *quotas.DynamicRateLimiter
+		retryPolicy      backoff.RetryPolicy
+
+		metricsClient metrics.Client
+		logger        log.Logger
+
+		// This is the batch size used by pull based RPC replicator.
+		fetchTasksBatchSize dynamicconfig.IntPropertyFnWithShardIDFilter
+	}
+)
+
+var _ TaskAckManager = (*taskAckManagerImpl)(nil)
+
+// NewTaskAckManager initializes a new replication task ack manager
+func NewTaskAckManager(
+	shard shard.Context,
+	executionCache *exec.Cache,
+) TaskAckManager {
+
+	config := shard.GetConfig()
+	rateLimiter := quotas.NewDynamicRateLimiter(func() float64 {
+		return config.ReplicationTaskGenerationQPS()
+	})
+	retryPolicy := backoff.NewExponentialRetryPolicy(100 * time.Millisecond)
+	retryPolicy.SetMaximumAttempts(config.ReplicatorReadTaskMaxRetryCount())
+	retryPolicy.SetBackoffCoefficient(1)
+
+	return &taskAckManagerImpl{
+		shard:               shard,
+		executionCache:      executionCache,
+		executionManager:    shard.GetExecutionManager(),
+		historyManager:      shard.GetHistoryManager(),
+		rateLimiter:         rateLimiter,
+		retryPolicy:         retryPolicy,
+		metricsClient:       shard.GetMetricsClient(),
+		logger:              shard.GetLogger().WithTags(tag.ComponentReplicationAckManager),
+		fetchTasksBatchSize: config.ReplicatorProcessorFetchTasksBatchSize,
+	}
+}
+
+func (t *taskAckManagerImpl) GetTask(
+	ctx ctx.Context,
+	taskInfo *replicator.ReplicationTaskInfo,
+) (*replicator.ReplicationTask, error) {
+	task := &persistence.ReplicationTaskInfo{
+		DomainID:     taskInfo.GetDomainID(),
+		WorkflowID:   taskInfo.GetWorkflowID(),
+		RunID:        taskInfo.GetRunID(),
+		TaskID:       taskInfo.GetTaskID(),
+		TaskType:     int(taskInfo.GetTaskType()),
+		FirstEventID: taskInfo.GetFirstEventID(),
+		NextEventID:  taskInfo.GetNextEventID(),
+		Version:      taskInfo.GetVersion(),
+		ScheduledID:  taskInfo.GetScheduledID(),
+	}
+	return t.toReplicationTask(ctx, task)
+}
+
+func (t *taskAckManagerImpl) GetTasks(
+	ctx ctx.Context,
+	pollingCluster string,
+	lastReadTaskID int64,
+) (*replicator.ReplicationMessages, error) {
+
+	if lastReadTaskID == common.EmptyMessageID {
+		lastReadTaskID = t.shard.GetClusterReplicationLevel(pollingCluster)
+	}
+
+	shardID := t.shard.GetShardID()
+	replicationScope := t.metricsClient.Scope(
+		metrics.ReplicatorQueueProcessorScope,
+		metrics.InstanceTag(strconv.Itoa(shardID)),
+	)
+	taskGeneratedTimer := replicationScope.StartTimer(metrics.TaskLatency)
+	taskInfoList, hasMore, err := t.readTasksWithBatchSize(lastReadTaskID, t.fetchTasksBatchSize(shardID))
+	if err != nil {
+		return nil, err
+	}
+
+	var replicationTasks []*replicator.ReplicationTask
+	readLevel := lastReadTaskID
+	for _, taskInfo := range taskInfoList {
+		_ = t.rateLimiter.Wait(ctx)
+		var replicationTask *replicator.ReplicationTask
+		op := func() error {
+			var err error
+			replicationTask, err = t.toReplicationTask(ctx, taskInfo)
+			return err
+		}
+
+		err = backoff.Retry(op, t.retryPolicy, common.IsPersistenceTransientError)
+		if err != nil {
+			t.logger.Debug("Failed to get replication task. Return what we have so far.", tag.Error(err))
+			hasMore = true
+			break
+		}
+		readLevel = taskInfo.GetTaskID()
+		if replicationTask != nil {
+			replicationTasks = append(replicationTasks, replicationTask)
+		}
+	}
+
+	taskGeneratedTimer.Stop()
+
+	replicationScope.RecordTimer(
+		metrics.ReplicationTasksLag,
+		time.Duration(t.shard.GetTransferMaxReadLevel()-readLevel),
+	)
+
+	replicationScope.RecordTimer(
+		metrics.ReplicationTasksFetched,
+		time.Duration(len(taskInfoList)),
+	)
+
+	replicationScope.RecordTimer(
+		metrics.ReplicationTasksReturned,
+		time.Duration(len(replicationTasks)),
+	)
+
+	if err := t.shard.UpdateClusterReplicationLevel(
+		pollingCluster,
+		lastReadTaskID,
+	); err != nil {
+		t.logger.Error("error updating replication level for shard", tag.Error(err), tag.OperationFailed)
+	}
+
+	return &replicator.ReplicationMessages{
+		ReplicationTasks:       replicationTasks,
+		HasMore:                common.BoolPtr(hasMore),
+		LastRetrievedMessageId: common.Int64Ptr(readLevel),
+	}, nil
+}
+
+func (t *taskAckManagerImpl) toReplicationTask(
+	ctx ctx.Context,
+	taskInfo task.Info,
+) (*replicator.ReplicationTask, error) {
+
+	task, ok := taskInfo.(*persistence.ReplicationTaskInfo)
+	if !ok {
+		return nil, errUnknownQueueTask
+	}
+
+	switch task.TaskType {
+	case persistence.ReplicationTaskTypeSyncActivity:
+		task, err := t.generateSyncActivityTask(ctx, task)
+		if task != nil {
+			task.SourceTaskId = common.Int64Ptr(taskInfo.GetTaskID())
+		}
+		return task, err
+	case persistence.ReplicationTaskTypeHistory:
+		task, err := t.generateHistoryReplicationTask(ctx, task)
+		if task != nil {
+			task.SourceTaskId = common.Int64Ptr(taskInfo.GetTaskID())
+		}
+		return task, err
+	case persistence.ReplicationTaskTypeFailoverMarker:
+		task := t.generateFailoverMarkerTask(task)
+		if task != nil {
+			task.SourceTaskId = common.Int64Ptr(taskInfo.GetTaskID())
+		}
+		return task, nil
+	default:
+		return nil, errUnknownReplicationTask
+	}
+}
+
+func (t *taskAckManagerImpl) processReplication(
+	ctx ctx.Context,
+	processTaskIfClosed bool,
+	taskInfo *persistence.ReplicationTaskInfo,
+	action func(
+		activityInfo *persistence.ActivityInfo,
+		versionHistories *persistence.VersionHistories,
+	) (*replicator.ReplicationTask, error),
+) (retReplicationTask *replicator.ReplicationTask, retError error) {
+
+	execution := shared.WorkflowExecution{
+		WorkflowId: common.StringPtr(taskInfo.GetWorkflowID()),
+		RunId:      common.StringPtr(taskInfo.GetRunID()),
+	}
+
+	context, release, err := t.executionCache.GetOrCreateWorkflowExecution(ctx, taskInfo.GetDomainID(), execution)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { release(retError) }()
+
+	msBuilder, err := context.LoadWorkflowExecution()
+	switch err.(type) {
+	case nil:
+		if !processTaskIfClosed && !msBuilder.IsWorkflowExecutionRunning() {
+			// workflow already finished, no need to process the replication task
+			return nil, nil
+		}
+
+		var targetVersionHistory *persistence.VersionHistories
+		versionHistories := msBuilder.GetVersionHistories()
+		if versionHistories != nil {
+			targetVersionHistory = msBuilder.GetVersionHistories().Duplicate()
+		}
+
+		var targetActivityInfo *persistence.ActivityInfo
+		if activityInfo, ok := msBuilder.GetActivityInfo(
+			taskInfo.ScheduledID,
+		); ok {
+			targetActivityInfo = exec.CopyActivityInfo(activityInfo)
+		}
+		release(nil)
+
+		return action(targetActivityInfo, targetVersionHistory)
+	case *shared.EntityNotExistsError:
+		return nil, nil
+	default:
+		return nil, err
+	}
+}
+
+func (t *taskAckManagerImpl) getEventsBlob(
+	branchToken []byte,
+	firstEventID int64,
+	nextEventID int64,
+) (*shared.DataBlob, error) {
+
+	var eventBatchBlobs []*persistence.DataBlob
+	var pageToken []byte
+	req := &persistence.ReadHistoryBranchRequest{
+		BranchToken:   branchToken,
+		MinEventID:    firstEventID,
+		MaxEventID:    nextEventID,
+		PageSize:      t.shard.GetConfig().ReplicationTaskProcessorReadHistoryBatchSize(),
+		NextPageToken: pageToken,
+		ShardID:       common.IntPtr(t.shard.GetShardID()),
+	}
+
+	for {
+		resp, err := t.historyManager.ReadRawHistoryBranch(req)
+		if err != nil {
+			return nil, err
+		}
+
+		req.NextPageToken = resp.NextPageToken
+		eventBatchBlobs = append(eventBatchBlobs, resp.HistoryEventBlobs...)
+
+		if len(req.NextPageToken) == 0 {
+			break
+		}
+	}
+
+	if len(eventBatchBlobs) != 1 {
+		return nil, &shared.InternalServiceError{
+			Message: "replicatorQueueProcessor encounter more than 1 NDC raw event batch",
+		}
+	}
+
+	return eventBatchBlobs[0].ToThrift(), nil
+}
+
+func (t *taskAckManagerImpl) isNewRunNDCEnabled(
+	ctx ctx.Context,
+	domainID string,
+	workflowID string,
+	runID string,
+) (isNDCWorkflow bool, retError error) {
+
+	context, release, err := t.executionCache.GetOrCreateWorkflowExecution(
+		ctx,
+		domainID,
+		shared.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	if err != nil {
+		return false, err
+	}
+	defer func() { release(retError) }()
+
+	mutableState, err := context.LoadWorkflowExecution()
+	if err != nil {
+		return false, err
+	}
+	return mutableState.GetVersionHistories() != nil, nil
+}
+
+func (t *taskAckManagerImpl) readTasksWithBatchSize(
+	readLevel int64,
+	batchSize int,
+) ([]task.Info, bool, error) {
+
+	response, err := t.executionManager.GetReplicationTasks(&persistence.GetReplicationTasksRequest{
+		ReadLevel:    readLevel,
+		MaxReadLevel: t.shard.GetTransferMaxReadLevel(),
+		BatchSize:    batchSize,
+	})
+
+	if err != nil {
+		return nil, false, err
+	}
+
+	tasks := make([]task.Info, len(response.Tasks))
+	for i := range response.Tasks {
+		tasks[i] = response.Tasks[i]
+	}
+
+	return tasks, len(response.NextPageToken) != 0, nil
+}
+
+func (t *taskAckManagerImpl) getAllHistory(
+	firstEventID int64,
+	nextEventID int64,
+	branchToken []byte,
+) (*shared.History, error) {
+
+	// overall result
+	shardID := t.shard.GetShardID()
+	var historyEvents []*shared.HistoryEvent
+	historySize := 0
+	iterator := collection.NewPagingIterator(
+		t.getPaginationFunc(
+			firstEventID,
+			nextEventID,
+			branchToken,
+			shardID,
+			&historySize,
+		),
+	)
+	for iterator.HasNext() {
+		event, err := iterator.Next()
+		if err != nil {
+			return nil, err
+		}
+		historyEvents = append(historyEvents, event.(*shared.HistoryEvent))
+	}
+	t.metricsClient.RecordTimer(metrics.ReplicatorQueueProcessorScope, metrics.HistorySize, time.Duration(historySize))
+	history := &shared.History{
+		Events: historyEvents,
+	}
+	return history, nil
+}
+
+func (t *taskAckManagerImpl) getPaginationFunc(
+	firstEventID int64,
+	nextEventID int64,
+	branchToken []byte,
+	shardID int,
+	historySize *int,
+) collection.PaginationFn {
+
+	return func(paginationToken []byte) ([]interface{}, []byte, error) {
+		events, _, pageToken, pageHistorySize, err := persistence.PaginateHistory(
+			t.historyManager,
+			false,
+			branchToken,
+			firstEventID,
+			nextEventID,
+			paginationToken,
+			defaultHistoryPageSize,
+			common.IntPtr(shardID),
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+		*historySize += pageHistorySize
+		var paginateItems []interface{}
+		for _, event := range events {
+			paginateItems = append(paginateItems, event)
+		}
+		return paginateItems, pageToken, nil
+	}
+}
+
+func (t *taskAckManagerImpl) generateReplicationTask(
+	targetClusters []string,
+	task *persistence.ReplicationTaskInfo,
+) (*replicator.ReplicationTask, string, error) {
+
+	history, err := t.getAllHistory(
+		task.FirstEventID,
+		task.NextEventID,
+		task.BranchToken,
+	)
+	if err != nil {
+		return nil, "", err
+	}
+	for _, event := range history.Events {
+		if task.Version != event.GetVersion() {
+			return nil, "", nil
+		}
+	}
+
+	var newRunID string
+	var newRunHistory *shared.History
+	events := history.Events
+	if len(events) > 0 {
+		lastEvent := events[len(events)-1]
+		if lastEvent.GetEventType() == shared.EventTypeWorkflowExecutionContinuedAsNew {
+			// Check if this is replication task for ContinueAsNew event, then retrieve the history for new execution
+			newRunID = lastEvent.WorkflowExecutionContinuedAsNewEventAttributes.GetNewExecutionRunId()
+			newRunHistory, err = t.getAllHistory(
+				common.FirstEventID,
+				common.FirstEventID+1, // [common.FirstEventID to common.FirstEventID+1) will get the first batch
+				task.NewRunBranchToken,
+			)
+			if err != nil {
+				return nil, "", err
+			}
+		}
+	}
+
+	ret := &replicator.ReplicationTask{
+		TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeHistory),
+		HistoryTaskAttributes: &replicator.HistoryTaskAttributes{
+			TargetClusters:  targetClusters,
+			DomainId:        common.StringPtr(task.DomainID),
+			WorkflowId:      common.StringPtr(task.WorkflowID),
+			RunId:           common.StringPtr(task.RunID),
+			FirstEventId:    common.Int64Ptr(task.FirstEventID),
+			NextEventId:     common.Int64Ptr(task.NextEventID),
+			Version:         common.Int64Ptr(task.Version),
+			ReplicationInfo: convertLastReplicationInfo(task.LastReplicationInfo),
+			History:         history,
+			NewRunHistory:   newRunHistory,
+			ResetWorkflow:   common.BoolPtr(task.ResetWorkflow),
+		},
+	}
+	return ret, newRunID, nil
+}
+
+func (t *taskAckManagerImpl) generateFailoverMarkerTask(
+	taskInfo *persistence.ReplicationTaskInfo,
+) *replicator.ReplicationTask {
+
+	return &replicator.ReplicationTask{
+		TaskType:     replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeFailoverMarker),
+		SourceTaskId: common.Int64Ptr(taskInfo.GetTaskID()),
+		FailoverMarkerAttributes: &replicator.FailoverMarkerAttributes{
+			DomainID:        common.StringPtr(taskInfo.GetDomainID()),
+			FailoverVersion: common.Int64Ptr(taskInfo.GetVersion()),
+			CreationTime:    common.Int64Ptr(taskInfo.CreationTime),
+		},
+	}
+}
+
+func (t *taskAckManagerImpl) generateSyncActivityTask(
+	ctx ctx.Context,
+	taskInfo *persistence.ReplicationTaskInfo,
+) (*replicator.ReplicationTask, error) {
+
+	return t.processReplication(
+		ctx,
+		false, // not necessary to send out sync activity task if workflow closed
+		taskInfo,
+		func(
+			activityInfo *persistence.ActivityInfo,
+			versionHistories *persistence.VersionHistories,
+		) (*replicator.ReplicationTask, error) {
+			if activityInfo == nil {
+				return nil, nil
+			}
+
+			var startedTime *int64
+			var heartbeatTime *int64
+			scheduledTime := common.Int64Ptr(activityInfo.ScheduledTime.UnixNano())
+			if activityInfo.StartedID != common.EmptyEventID {
+				startedTime = common.Int64Ptr(activityInfo.StartedTime.UnixNano())
+			}
+			// LastHeartBeatUpdatedTime must be valid when getting the sync activity replication task
+			heartbeatTime = common.Int64Ptr(activityInfo.LastHeartBeatUpdatedTime.UnixNano())
+
+			//Version history uses when replicate the sync activity task
+			var versionHistory *shared.VersionHistory
+			if versionHistories != nil {
+				rawVersionHistory, err := versionHistories.GetCurrentVersionHistory()
+				if err != nil {
+					return nil, err
+				}
+				versionHistory = rawVersionHistory.ToThrift()
+			}
+
+			return &replicator.ReplicationTask{
+				TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeSyncActivity),
+				SyncActivityTaskAttributes: &replicator.SyncActivityTaskAttributes{
+					DomainId:           common.StringPtr(taskInfo.GetDomainID()),
+					WorkflowId:         common.StringPtr(taskInfo.GetWorkflowID()),
+					RunId:              common.StringPtr(taskInfo.GetRunID()),
+					Version:            common.Int64Ptr(activityInfo.Version),
+					ScheduledId:        common.Int64Ptr(activityInfo.ScheduleID),
+					ScheduledTime:      scheduledTime,
+					StartedId:          common.Int64Ptr(activityInfo.StartedID),
+					StartedTime:        startedTime,
+					LastHeartbeatTime:  heartbeatTime,
+					Details:            activityInfo.Details,
+					Attempt:            common.Int32Ptr(activityInfo.Attempt),
+					LastFailureReason:  common.StringPtr(activityInfo.LastFailureReason),
+					LastWorkerIdentity: common.StringPtr(activityInfo.LastWorkerIdentity),
+					LastFailureDetails: activityInfo.LastFailureDetails,
+					VersionHistory:     versionHistory,
+				},
+			}, nil
+		},
+	)
+}
+
+func (t *taskAckManagerImpl) generateHistoryReplicationTask(
+	ctx ctx.Context,
+	task *persistence.ReplicationTaskInfo,
+) (*replicator.ReplicationTask, error) {
+
+	return t.processReplication(
+		ctx,
+		true, // still necessary to send out history replication message if workflow closed
+		task,
+		func(
+			activityInfo *persistence.ActivityInfo,
+			versionHistories *persistence.VersionHistories,
+		) (*replicator.ReplicationTask, error) {
+			// TODO when 3+DC migration is done, remove this block of code
+			if versionHistories == nil {
+				domainEntry, err := t.shard.GetDomainCache().GetDomainByID(task.DomainID)
+				if err != nil {
+					return nil, err
+				}
+
+				var targetClusters []string
+				for _, cluster := range domainEntry.GetReplicationConfig().Clusters {
+					targetClusters = append(targetClusters, cluster.ClusterName)
+				}
+
+				replicationTask, newRunID, err := t.generateReplicationTask(
+					targetClusters,
+					task,
+				)
+				if err != nil {
+					return nil, err
+				}
+				if newRunID != "" {
+					isNDCWorkflow, err := t.isNewRunNDCEnabled(ctx, task.DomainID, task.WorkflowID, newRunID)
+					if err != nil {
+						return nil, err
+					}
+					replicationTask.HistoryTaskAttributes.NewRunNDC = common.BoolPtr(isNDCWorkflow)
+				}
+
+				return replicationTask, err
+			}
+
+			// NDC workflow
+			versionHistoryItems, branchToken, err := getVersionHistoryItems(
+				versionHistories,
+				task.FirstEventID,
+				task.Version,
+			)
+			if err != nil {
+				return nil, err
+			}
+
+			// BranchToken will not set in get dlq replication message request
+			if len(task.BranchToken) == 0 {
+				task.BranchToken = branchToken
+			}
+
+			eventsBlob, err := t.getEventsBlob(
+				task.BranchToken,
+				task.FirstEventID,
+				task.NextEventID,
+			)
+			if err != nil {
+				return nil, err
+			}
+
+			var newRunEventsBlob *shared.DataBlob
+			if len(task.NewRunBranchToken) != 0 {
+				// only get the first batch
+				newRunEventsBlob, err = t.getEventsBlob(
+					task.NewRunBranchToken,
+					common.FirstEventID,
+					common.FirstEventID+1,
+				)
+				if err != nil {
+					return nil, err
+				}
+			}
+
+			replicationTask := &replicator.ReplicationTask{
+				TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeHistoryV2),
+				HistoryTaskV2Attributes: &replicator.HistoryTaskV2Attributes{
+					TaskId:              common.Int64Ptr(task.FirstEventID),
+					DomainId:            common.StringPtr(task.DomainID),
+					WorkflowId:          common.StringPtr(task.WorkflowID),
+					RunId:               common.StringPtr(task.RunID),
+					VersionHistoryItems: versionHistoryItems,
+					Events:              eventsBlob,
+					NewRunEvents:        newRunEventsBlob,
+				},
+			}
+			return replicationTask, nil
+		},
+	)
+}
+
+func getVersionHistoryItems(
+	versionHistories *persistence.VersionHistories,
+	eventID int64,
+	version int64,
+) ([]*shared.VersionHistoryItem, []byte, error) {
+
+	if versionHistories == nil {
+		return nil, nil, &shared.InternalServiceError{
+			Message: "replicatorQueueProcessor encounter workflow without version histories",
+		}
+	}
+
+	versionHistoryIndex, err := versionHistories.FindFirstVersionHistoryIndexByItem(
+		persistence.NewVersionHistoryItem(
+			eventID,
+			version,
+		),
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	versionHistory, err := versionHistories.GetVersionHistory(versionHistoryIndex)
+	if err != nil {
+		return nil, nil, err
+	}
+	return versionHistory.ToThrift().Items, versionHistory.GetBranchToken(), nil
+}
+
+// TODO deprecate when 3+DC is released
+func convertLastReplicationInfo(info map[string]*persistence.ReplicationInfo) map[string]*shared.ReplicationInfo {
+	replicationInfoMap := make(map[string]*shared.ReplicationInfo)
+	for k, v := range info {
+		replicationInfoMap[k] = &shared.ReplicationInfo{
+			Version:     common.Int64Ptr(v.Version),
+			LastEventId: common.Int64Ptr(v.LastEventID),
+		}
+	}
+
+	return replicationInfoMap
+}
diff --git a/service/history/replication/task_ack_manager_mock.go b/service/history/replication/task_ack_manager_mock.go
new file mode 100644
index 000000000..b57c0e12f
--- /dev/null
+++ b/service/history/replication/task_ack_manager_mock.go
@@ -0,0 +1,90 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+//
+
+// Code generated by MockGen. DO NOT EDIT.
+// Source: task_ack_manager.go
+
+// Package replication is a generated GoMock package.
+package replication
+
+import (
+	context "context"
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+
+	replicator "github.com/uber/cadence/.gen/go/replicator"
+)
+
+// MockTaskAckManager is a mock of TaskAckManager interface
+type MockTaskAckManager struct {
+	ctrl     *gomock.Controller
+	recorder *MockTaskAckManagerMockRecorder
+}
+
+// MockTaskAckManagerMockRecorder is the mock recorder for MockTaskAckManager
+type MockTaskAckManagerMockRecorder struct {
+	mock *MockTaskAckManager
+}
+
+// NewMockTaskAckManager creates a new mock instance
+func NewMockTaskAckManager(ctrl *gomock.Controller) *MockTaskAckManager {
+	mock := &MockTaskAckManager{ctrl: ctrl}
+	mock.recorder = &MockTaskAckManagerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockTaskAckManager) EXPECT() *MockTaskAckManagerMockRecorder {
+	return m.recorder
+}
+
+// GetTask mocks base method
+func (m *MockTaskAckManager) GetTask(ctx context.Context, taskInfo *replicator.ReplicationTaskInfo) (*replicator.ReplicationTask, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetTask", ctx, taskInfo)
+	ret0, _ := ret[0].(*replicator.ReplicationTask)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetTask indicates an expected call of GetTask
+func (mr *MockTaskAckManagerMockRecorder) GetTask(ctx, taskInfo interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTask", reflect.TypeOf((*MockTaskAckManager)(nil).GetTask), ctx, taskInfo)
+}
+
+// GetTasks mocks base method
+func (m *MockTaskAckManager) GetTasks(ctx context.Context, pollingCluster string, lastReadTaskID int64) (*replicator.ReplicationMessages, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetTasks", ctx, pollingCluster, lastReadTaskID)
+	ret0, _ := ret[0].(*replicator.ReplicationMessages)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetTasks indicates an expected call of GetTasks
+func (mr *MockTaskAckManagerMockRecorder) GetTasks(ctx, pollingCluster, lastReadTaskID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTasks", reflect.TypeOf((*MockTaskAckManager)(nil).GetTasks), ctx, pollingCluster, lastReadTaskID)
+}
diff --git a/service/history/replication/task_fetcher.go b/service/history/replication/task_fetcher.go
index 53ac1a8fb..ab3aa4d48 100644
--- a/service/history/replication/task_fetcher.go
+++ b/service/history/replication/task_fetcher.go
@@ -37,7 +37,6 @@ import (
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/quotas"
-	serviceConfig "github.com/uber/cadence/common/service/config"
 	"github.com/uber/cadence/service/history/config"
 )
 
@@ -91,30 +90,27 @@ var _ TaskFetchers = (*taskFetchersImpl)(nil)
 func NewTaskFetchers(
 	logger log.Logger,
 	config *config.Config,
-	consumerConfig *serviceConfig.ReplicationConsumerConfig,
 	clusterMetadata cluster.Metadata,
 	clientBean client.Bean,
 ) TaskFetchers {
 
 	var fetchers []TaskFetcher
-	if consumerConfig.Type == serviceConfig.ReplicationConsumerTypeRPC {
-		for clusterName, info := range clusterMetadata.GetAllClusterInfo() {
-			if !info.Enabled {
-				continue
-			}
+	for clusterName, info := range clusterMetadata.GetAllClusterInfo() {
+		if !info.Enabled {
+			continue
+		}
 
-			currentCluster := clusterMetadata.GetCurrentClusterName()
-			if clusterName != currentCluster {
-				remoteFrontendClient := clientBean.GetRemoteAdminClient(clusterName)
-				fetcher := newReplicationTaskFetcher(
-					logger,
-					clusterName,
-					currentCluster,
-					config,
-					remoteFrontendClient,
-				)
-				fetchers = append(fetchers, fetcher)
-			}
+		currentCluster := clusterMetadata.GetCurrentClusterName()
+		if clusterName != currentCluster {
+			remoteFrontendClient := clientBean.GetRemoteAdminClient(clusterName)
+			fetcher := newReplicationTaskFetcher(
+				logger,
+				clusterName,
+				currentCluster,
+				config,
+				remoteFrontendClient,
+			)
+			fetchers = append(fetchers, fetcher)
 		}
 	}
 
diff --git a/service/history/replicatorQueueProcessor.go b/service/history/replicatorQueueProcessor.go
deleted file mode 100644
index 0bb9f56cf..000000000
--- a/service/history/replicatorQueueProcessor.go
+++ /dev/null
@@ -1,896 +0,0 @@
-// Copyright (c) 2017 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-//TODO move this file to replication subfolder
-
-package history
-
-import (
-	ctx "context"
-	"errors"
-	"strconv"
-	"time"
-
-	"github.com/uber/cadence/.gen/go/replicator"
-	"github.com/uber/cadence/.gen/go/shared"
-	"github.com/uber/cadence/common"
-	"github.com/uber/cadence/common/backoff"
-	"github.com/uber/cadence/common/clock"
-	"github.com/uber/cadence/common/log"
-	"github.com/uber/cadence/common/log/tag"
-	"github.com/uber/cadence/common/messaging"
-	"github.com/uber/cadence/common/metrics"
-	"github.com/uber/cadence/common/persistence"
-	"github.com/uber/cadence/common/quotas"
-	"github.com/uber/cadence/common/service/dynamicconfig"
-	"github.com/uber/cadence/service/history/execution"
-	"github.com/uber/cadence/service/history/shard"
-	"github.com/uber/cadence/service/history/task"
-)
-
-type (
-	replicatorQueueProcessorImpl struct {
-		currentClusterName    string
-		shard                 shard.Context
-		executionCache        *execution.Cache
-		replicationTaskFilter task.Filter
-		executionMgr          persistence.ExecutionManager
-		historyV2Mgr          persistence.HistoryManager
-		replicator            messaging.Producer
-		metricsClient         metrics.Client
-		options               *QueueProcessorOptions
-		logger                log.Logger
-		retryPolicy           backoff.RetryPolicy
-		// This is the batch size used by pull based RPC replicator.
-		fetchTasksBatchSize dynamicconfig.IntPropertyFnWithShardIDFilter
-		*queueProcessorBase
-		queueAckMgr
-
-		lastShardSyncTimestamp time.Time
-		rateLimiter            *quotas.DynamicRateLimiter
-	}
-)
-
-var (
-	errUnknownReplicationTask = errors.New("unknown replication task")
-	errHistoryNotFoundTask    = errors.New("history not found")
-	defaultHistoryPageSize    = 1000
-)
-
-func newReplicatorQueueProcessor(
-	shard shard.Context,
-	executionCache *execution.Cache,
-	replicator messaging.Producer,
-	executionMgr persistence.ExecutionManager,
-	historyV2Mgr persistence.HistoryManager,
-	logger log.Logger,
-) ReplicatorQueueProcessor {
-
-	currentClusterName := shard.GetService().GetClusterMetadata().GetCurrentClusterName()
-
-	config := shard.GetConfig()
-	options := &QueueProcessorOptions{
-		BatchSize:                           config.ReplicatorTaskBatchSize,
-		WorkerCount:                         config.ReplicatorTaskWorkerCount,
-		MaxPollRPS:                          config.ReplicatorProcessorMaxPollRPS,
-		MaxPollInterval:                     config.ReplicatorProcessorMaxPollInterval,
-		MaxPollIntervalJitterCoefficient:    config.ReplicatorProcessorMaxPollIntervalJitterCoefficient,
-		UpdateAckInterval:                   config.ReplicatorProcessorUpdateAckInterval,
-		UpdateAckIntervalJitterCoefficient:  config.ReplicatorProcessorUpdateAckIntervalJitterCoefficient,
-		MaxRetryCount:                       config.ReplicatorTaskMaxRetryCount,
-		RedispatchInterval:                  config.StandbyTaskRedispatchInterval,
-		RedispatchIntervalJitterCoefficient: config.TaskRedispatchIntervalJitterCoefficient,
-		MaxRedispatchQueueSize:              config.ReplicatorProcessorMaxRedispatchQueueSize,
-		EnablePriorityTaskProcessor:         config.ReplicatorProcessorEnablePriorityTaskProcessor,
-		MetricScope:                         metrics.ReplicatorQueueProcessorScope,
-		QueueType:                           task.QueueTypeReplication,
-	}
-
-	logger = logger.WithTags(tag.ComponentReplicatorQueue)
-
-	replicationTaskFilter := func(taskInfo task.Info) (bool, error) {
-		return true, nil
-	}
-
-	retryPolicy := backoff.NewExponentialRetryPolicy(100 * time.Millisecond)
-	retryPolicy.SetMaximumAttempts(config.ReplicatorReadTaskMaxRetryCount())
-	retryPolicy.SetBackoffCoefficient(1)
-
-	processor := &replicatorQueueProcessorImpl{
-		currentClusterName:    currentClusterName,
-		shard:                 shard,
-		executionCache:        executionCache,
-		replicationTaskFilter: replicationTaskFilter,
-		executionMgr:          executionMgr,
-		historyV2Mgr:          historyV2Mgr,
-		replicator:            replicator,
-		metricsClient:         shard.GetMetricsClient(),
-		options:               options,
-		logger:                logger,
-		retryPolicy:           retryPolicy,
-		fetchTasksBatchSize:   config.ReplicatorProcessorFetchTasksBatchSize,
-	}
-
-	queueAckMgr := newQueueAckMgr(shard, options, processor, shard.GetReplicatorAckLevel(), logger)
-	queueProcessorBase := newQueueProcessorBase(
-		currentClusterName,
-		shard,
-		options,
-		processor,
-		nil, // replicator queue processor will soon be deprecated and won't use priority task processor
-		queueAckMgr,
-		executionCache,
-		replicationTaskFilter,
-		nil, // there's no queueTask implementation for replication task
-		logger,
-		shard.GetMetricsClient().Scope(metrics.ReplicatorQueueProcessorScope),
-	)
-	processor.queueAckMgr = queueAckMgr
-	processor.queueProcessorBase = queueProcessorBase
-	processor.rateLimiter = quotas.NewDynamicRateLimiter(func() float64 {
-		return config.ReplicationTaskGenerationQPS()
-	})
-	return processor
-}
-
-func (p *replicatorQueueProcessorImpl) getTaskFilter() task.Filter {
-	return p.replicationTaskFilter
-}
-
-func (p *replicatorQueueProcessorImpl) complete(
-	taskInfo *taskInfo,
-) {
-	p.queueProcessorBase.complete(taskInfo.task)
-}
-
-func (p *replicatorQueueProcessorImpl) process(
-	taskInfo *taskInfo,
-) (int, error) {
-
-	task, ok := taskInfo.task.(*persistence.ReplicationTaskInfo)
-	if !ok {
-		return metrics.ReplicatorQueueProcessorScope, errUnexpectedQueueTask
-	}
-	// replication queue should always process all tasks
-	// so should not do anything to shouldProcessTask variable
-
-	switch task.TaskType {
-	case persistence.ReplicationTaskTypeSyncActivity:
-		err := p.processSyncActivityTask(task)
-		if err == nil {
-			err = p.executionMgr.CompleteReplicationTask(&persistence.CompleteReplicationTaskRequest{TaskID: task.GetTaskID()})
-		}
-		return metrics.ReplicatorTaskSyncActivityScope, err
-	case persistence.ReplicationTaskTypeHistory:
-		err := p.processHistoryReplicationTask(task)
-		if _, ok := err.(*shared.EntityNotExistsError); ok {
-			err = errHistoryNotFoundTask
-		}
-		if err == nil {
-			err = p.executionMgr.CompleteReplicationTask(&persistence.CompleteReplicationTaskRequest{TaskID: task.GetTaskID()})
-		}
-		return metrics.ReplicatorTaskHistoryScope, err
-	default:
-		return metrics.ReplicatorQueueProcessorScope, errUnknownReplicationTask
-	}
-}
-
-func (p *replicatorQueueProcessorImpl) queueShutdown() error {
-	// there is no shutdown specific behavior for replication queue
-	return nil
-}
-
-func (p *replicatorQueueProcessorImpl) processSyncActivityTask(
-	task *persistence.ReplicationTaskInfo,
-) error {
-
-	replicationTask, err := p.generateSyncActivityTask(ctx.Background(), task)
-	if err != nil || replicationTask == nil {
-		return err
-	}
-
-	return p.replicator.Publish(replicationTask)
-}
-
-func (p *replicatorQueueProcessorImpl) processHistoryReplicationTask(
-	task *persistence.ReplicationTaskInfo,
-) error {
-
-	replicationTask, err := p.toReplicationTask(ctx.Background(), task)
-	if err != nil || replicationTask == nil {
-		return err
-	}
-
-	err = p.replicator.Publish(replicationTask)
-	if err == messaging.ErrMessageSizeLimit {
-		// message size exceeds the server messaging size limit
-		// for this specific case, just send out a metadata message and
-		// let receiver fetch from source (for the concrete history events)
-		if metadataTask := p.generateHistoryMetadataTask(
-			task,
-			replicationTask,
-		); metadataTask != nil {
-			err = p.replicator.Publish(metadataTask)
-		}
-	}
-	return err
-}
-
-func (p *replicatorQueueProcessorImpl) generateHistoryMetadataTask(
-	task *persistence.ReplicationTaskInfo,
-	replicationTask *replicator.ReplicationTask,
-) *replicator.ReplicationTask {
-
-	if replicationTask.HistoryTaskAttributes != nil {
-		return &replicator.ReplicationTask{
-			TaskType: replicator.ReplicationTaskTypeHistoryMetadata.Ptr(),
-			HistoryMetadataTaskAttributes: &replicator.HistoryMetadataTaskAttributes{
-				TargetClusters: replicationTask.HistoryTaskAttributes.TargetClusters,
-				DomainId:       common.StringPtr(task.DomainID),
-				WorkflowId:     common.StringPtr(task.WorkflowID),
-				RunId:          common.StringPtr(task.RunID),
-				FirstEventId:   common.Int64Ptr(task.FirstEventID),
-				NextEventId:    common.Int64Ptr(task.NextEventID),
-			},
-		}
-	} else if replicationTask.HistoryTaskV2Attributes != nil {
-		return &replicator.ReplicationTask{
-			TaskType: replicator.ReplicationTaskTypeHistoryMetadata.Ptr(),
-			HistoryMetadataTaskAttributes: &replicator.HistoryMetadataTaskAttributes{
-				DomainId:     common.StringPtr(task.DomainID),
-				WorkflowId:   common.StringPtr(task.WorkflowID),
-				RunId:        common.StringPtr(task.RunID),
-				FirstEventId: common.Int64Ptr(task.FirstEventID),
-				NextEventId:  common.Int64Ptr(task.NextEventID),
-				Version:      common.Int64Ptr(task.Version),
-			},
-		}
-	}
-	return nil
-}
-
-// GenerateReplicationTask generate replication task
-func GenerateReplicationTask(
-	targetClusters []string,
-	task *persistence.ReplicationTaskInfo,
-	historyV2Mgr persistence.HistoryManager,
-	metricsClient metrics.Client,
-	history *shared.History,
-	shardID *int,
-) (*replicator.ReplicationTask, string, error) {
-	var err error
-	if history == nil {
-		history, _, err = GetAllHistory(historyV2Mgr, metricsClient, false,
-			task.FirstEventID, task.NextEventID, task.BranchToken, shardID)
-		if err != nil {
-			return nil, "", err
-		}
-		for _, event := range history.Events {
-			if task.Version != event.GetVersion() {
-				return nil, "", nil
-			}
-		}
-	}
-
-	var newRunID string
-	var newRunHistory *shared.History
-	events := history.Events
-	if len(events) > 0 {
-		lastEvent := events[len(events)-1]
-		if lastEvent.GetEventType() == shared.EventTypeWorkflowExecutionContinuedAsNew {
-			// Check if this is replication task for ContinueAsNew event, then retrieve the history for new execution
-			newRunID = lastEvent.WorkflowExecutionContinuedAsNewEventAttributes.GetNewExecutionRunId()
-			newRunHistory, _, err = GetAllHistory(
-				historyV2Mgr,
-				metricsClient,
-				false,
-				common.FirstEventID,
-				common.FirstEventID+1, // [common.FirstEventID to common.FirstEventID+1) will get the first batch
-				task.NewRunBranchToken,
-				shardID)
-			if err != nil {
-				return nil, "", err
-			}
-		}
-	}
-
-	ret := &replicator.ReplicationTask{
-		TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeHistory),
-		HistoryTaskAttributes: &replicator.HistoryTaskAttributes{
-			TargetClusters:  targetClusters,
-			DomainId:        common.StringPtr(task.DomainID),
-			WorkflowId:      common.StringPtr(task.WorkflowID),
-			RunId:           common.StringPtr(task.RunID),
-			FirstEventId:    common.Int64Ptr(task.FirstEventID),
-			NextEventId:     common.Int64Ptr(task.NextEventID),
-			Version:         common.Int64Ptr(task.Version),
-			ReplicationInfo: convertLastReplicationInfo(task.LastReplicationInfo),
-			History:         history,
-			NewRunHistory:   newRunHistory,
-			ResetWorkflow:   common.BoolPtr(task.ResetWorkflow),
-		},
-	}
-	return ret, newRunID, nil
-}
-
-func (p *replicatorQueueProcessorImpl) readTasks(readLevel int64) ([]task.Info, bool, error) {
-	return p.readTasksWithBatchSize(readLevel, p.options.BatchSize())
-}
-
-func (p *replicatorQueueProcessorImpl) updateAckLevel(ackLevel int64) error {
-	err := p.shard.UpdateReplicatorAckLevel(ackLevel)
-
-	// TODO: Remove this after enabled the rpc replication
-	clusterMetadata := p.shard.GetClusterMetadata()
-	for name, cluster := range clusterMetadata.GetAllClusterInfo() {
-		if !cluster.Enabled || clusterMetadata.GetCurrentClusterName() == name {
-			continue
-		}
-		p.shard.UpdateClusterReplicationLevel(name, ackLevel)
-	}
-
-	// this is a hack, since there is not dedicated ticker on the queue processor
-	// to periodically send out sync shard message, put it here
-	now := clock.NewRealTimeSource().Now()
-	if p.lastShardSyncTimestamp.Add(p.shard.GetConfig().ShardSyncMinInterval()).Before(now) {
-		syncStatusTask := &replicator.ReplicationTask{
-			TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeSyncShardStatus),
-			SyncShardStatusTaskAttributes: &replicator.SyncShardStatusTaskAttributes{
-				SourceCluster: common.StringPtr(p.currentClusterName),
-				ShardId:       common.Int64Ptr(int64(p.shard.GetShardID())),
-				Timestamp:     common.Int64Ptr(now.UnixNano()),
-			},
-		}
-		// ignore the error
-		if syncErr := p.replicator.Publish(syncStatusTask); syncErr == nil {
-			p.lastShardSyncTimestamp = now
-		}
-	}
-	return err
-}
-
-// GetAllHistory return history
-func GetAllHistory(
-	historyV2Mgr persistence.HistoryManager,
-	metricsClient metrics.Client,
-	byBatch bool,
-	firstEventID int64,
-	nextEventID int64,
-	branchToken []byte,
-	shardID *int,
-) (*shared.History, []*shared.History, error) {
-
-	// overall result
-	var historyEvents []*shared.HistoryEvent
-	var historyBatches []*shared.History
-	historySize := 0
-	var err error
-
-	// variable used for each page
-	var pageHistoryEvents []*shared.HistoryEvent
-	var pageHistoryBatches []*shared.History
-	var pageToken []byte
-	var pageHistorySize int
-
-	for hasMore := true; hasMore; hasMore = len(pageToken) > 0 {
-		pageHistoryEvents, pageHistoryBatches, pageToken, pageHistorySize, err = persistence.PaginateHistory(
-			historyV2Mgr, byBatch,
-			branchToken, firstEventID, nextEventID,
-			pageToken, defaultHistoryPageSize, shardID,
-		)
-		if err != nil {
-			return nil, nil, err
-		}
-
-		historyEvents = append(historyEvents, pageHistoryEvents...)
-		historyBatches = append(historyBatches, pageHistoryBatches...)
-		historySize += pageHistorySize
-	}
-
-	// Emit metric and log for history size
-	if metricsClient != nil {
-		metricsClient.RecordTimer(metrics.ReplicatorQueueProcessorScope, metrics.HistorySize, time.Duration(historySize))
-	}
-
-	history := &shared.History{
-		Events: historyEvents,
-	}
-	return history, historyBatches, nil
-}
-
-// TODO deprecate when 3+DC is released
-func convertLastReplicationInfo(info map[string]*persistence.ReplicationInfo) map[string]*shared.ReplicationInfo {
-	replicationInfoMap := make(map[string]*shared.ReplicationInfo)
-	for k, v := range info {
-		replicationInfoMap[k] = &shared.ReplicationInfo{
-			Version:     common.Int64Ptr(v.Version),
-			LastEventId: common.Int64Ptr(v.LastEventID),
-		}
-	}
-
-	return replicationInfoMap
-}
-
-// TODO: when kafka deprecation is finished, delete all logic above
-//  and move logic below to dedicated replicationTaskAckMgr
-
-func (p *replicatorQueueProcessorImpl) getTasks(
-	ctx ctx.Context,
-	pollingCluster string,
-	lastReadTaskID int64,
-) (*replicator.ReplicationMessages, error) {
-
-	if lastReadTaskID == common.EmptyMessageID {
-		lastReadTaskID = p.shard.GetClusterReplicationLevel(pollingCluster)
-	}
-
-	replicationScope := p.metricsClient.Scope(
-		metrics.ReplicatorQueueProcessorScope,
-		metrics.InstanceTag(strconv.Itoa(p.shard.GetShardID())),
-	)
-	taskGeneratedTimer := replicationScope.StartTimer(metrics.TaskLatency)
-	shardID := p.shard.GetShardID()
-	taskInfoList, hasMore, err := p.readTasksWithBatchSize(lastReadTaskID, p.fetchTasksBatchSize(shardID))
-	if err != nil {
-		return nil, err
-	}
-
-	var replicationTasks []*replicator.ReplicationTask
-	readLevel := lastReadTaskID
-	for _, taskInfo := range taskInfoList {
-		_ = p.rateLimiter.Wait(ctx)
-		var replicationTask *replicator.ReplicationTask
-		op := func() error {
-			var err error
-			replicationTask, err = p.toReplicationTask(ctx, taskInfo)
-			return err
-		}
-
-		err = backoff.Retry(op, p.retryPolicy, common.IsPersistenceTransientError)
-		if err != nil {
-			p.logger.Debug("Failed to get replication task. Return what we have so far.", tag.Error(err))
-			hasMore = true
-			break
-		}
-		readLevel = taskInfo.GetTaskID()
-		if replicationTask != nil {
-			replicationTasks = append(replicationTasks, replicationTask)
-		}
-	}
-
-	taskGeneratedTimer.Stop()
-
-	replicationScope.RecordTimer(
-		metrics.ReplicationTasksLag,
-		time.Duration(p.shard.GetTransferMaxReadLevel()-readLevel),
-	)
-
-	replicationScope.RecordTimer(
-		metrics.ReplicationTasksFetched,
-		time.Duration(len(taskInfoList)),
-	)
-
-	replicationScope.RecordTimer(
-		metrics.ReplicationTasksReturned,
-		time.Duration(len(replicationTasks)),
-	)
-
-	if err := p.shard.UpdateClusterReplicationLevel(
-		pollingCluster,
-		lastReadTaskID,
-	); err != nil {
-		p.logger.Error("error updating replication level for shard", tag.Error(err), tag.OperationFailed)
-	}
-
-	if err := p.shard.UpdateReplicatorAckLevel(
-		lastReadTaskID,
-	); err != nil {
-		p.logger.Error("error updating kafka replication level", tag.Error(err), tag.OperationFailed)
-	}
-
-	return &replicator.ReplicationMessages{
-		ReplicationTasks:       replicationTasks,
-		HasMore:                common.BoolPtr(hasMore),
-		LastRetrievedMessageId: common.Int64Ptr(readLevel),
-	}, nil
-}
-
-func (p *replicatorQueueProcessorImpl) getTask(
-	ctx ctx.Context,
-	taskInfo *replicator.ReplicationTaskInfo,
-) (*replicator.ReplicationTask, error) {
-
-	task := &persistence.ReplicationTaskInfo{
-		DomainID:     taskInfo.GetDomainID(),
-		WorkflowID:   taskInfo.GetWorkflowID(),
-		RunID:        taskInfo.GetRunID(),
-		TaskID:       taskInfo.GetTaskID(),
-		TaskType:     int(taskInfo.GetTaskType()),
-		FirstEventID: taskInfo.GetFirstEventID(),
-		NextEventID:  taskInfo.GetNextEventID(),
-		Version:      taskInfo.GetVersion(),
-		ScheduledID:  taskInfo.GetScheduledID(),
-	}
-	return p.toReplicationTask(ctx, task)
-}
-
-func (p *replicatorQueueProcessorImpl) readTasksWithBatchSize(readLevel int64, batchSize int) ([]task.Info, bool, error) {
-	response, err := p.executionMgr.GetReplicationTasks(&persistence.GetReplicationTasksRequest{
-		ReadLevel:    readLevel,
-		MaxReadLevel: p.shard.GetTransferMaxReadLevel(),
-		BatchSize:    batchSize,
-	})
-
-	if err != nil {
-		return nil, false, err
-	}
-
-	tasks := make([]task.Info, len(response.Tasks))
-	for i := range response.Tasks {
-		tasks[i] = response.Tasks[i]
-	}
-
-	return tasks, len(response.NextPageToken) != 0, nil
-}
-
-func (p *replicatorQueueProcessorImpl) toReplicationTask(
-	ctx ctx.Context,
-	qTask task.Info,
-) (*replicator.ReplicationTask, error) {
-
-	task, ok := qTask.(*persistence.ReplicationTaskInfo)
-	if !ok {
-		return nil, errUnexpectedQueueTask
-	}
-
-	switch task.TaskType {
-	case persistence.ReplicationTaskTypeSyncActivity:
-		task, err := p.generateSyncActivityTask(ctx, task)
-		if task != nil {
-			task.SourceTaskId = common.Int64Ptr(qTask.GetTaskID())
-		}
-		return task, err
-	case persistence.ReplicationTaskTypeHistory:
-		task, err := p.generateHistoryReplicationTask(ctx, task)
-		if task != nil {
-			task.SourceTaskId = common.Int64Ptr(qTask.GetTaskID())
-		}
-		return task, err
-	case persistence.ReplicationTaskTypeFailoverMarker:
-		task := p.generateFailoverMarkerTask(task)
-		if task != nil {
-			task.SourceTaskId = common.Int64Ptr(qTask.GetTaskID())
-		}
-		return task, nil
-	default:
-		return nil, errUnknownReplicationTask
-	}
-}
-
-func (p *replicatorQueueProcessorImpl) generateSyncActivityTask(
-	ctx ctx.Context,
-	taskInfo *persistence.ReplicationTaskInfo,
-) (*replicator.ReplicationTask, error) {
-
-	return p.processReplication(
-		ctx,
-		false, // not necessary to send out sync activity task if workflow closed
-		taskInfo.GetDomainID(),
-		taskInfo.GetWorkflowID(),
-		taskInfo.GetRunID(),
-		func(mutableState execution.MutableState) (*replicator.ReplicationTask, error) {
-			activityInfo, ok := mutableState.GetActivityInfo(taskInfo.ScheduledID)
-			if !ok {
-				return nil, nil
-			}
-
-			var startedTime *int64
-			var heartbeatTime *int64
-			scheduledTime := common.Int64Ptr(activityInfo.ScheduledTime.UnixNano())
-			if activityInfo.StartedID != common.EmptyEventID {
-				startedTime = common.Int64Ptr(activityInfo.StartedTime.UnixNano())
-			}
-			// LastHeartBeatUpdatedTime must be valid when getting the sync activity replication task
-			heartbeatTime = common.Int64Ptr(activityInfo.LastHeartBeatUpdatedTime.UnixNano())
-
-			//Version history uses when replicate the sync activity task
-			versionHistories := mutableState.GetVersionHistories()
-			var versionHistory *shared.VersionHistory
-			if versionHistories != nil {
-				rawVersionHistory, err := versionHistories.GetCurrentVersionHistory()
-				if err != nil {
-					return nil, err
-				}
-				versionHistory = rawVersionHistory.ToThrift()
-			}
-
-			return &replicator.ReplicationTask{
-				TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeSyncActivity),
-				SyncActivityTaskAttributes: &replicator.SyncActivityTaskAttributes{
-					DomainId:           common.StringPtr(taskInfo.GetDomainID()),
-					WorkflowId:         common.StringPtr(taskInfo.GetWorkflowID()),
-					RunId:              common.StringPtr(taskInfo.GetRunID()),
-					Version:            common.Int64Ptr(activityInfo.Version),
-					ScheduledId:        common.Int64Ptr(activityInfo.ScheduleID),
-					ScheduledTime:      scheduledTime,
-					StartedId:          common.Int64Ptr(activityInfo.StartedID),
-					StartedTime:        startedTime,
-					LastHeartbeatTime:  heartbeatTime,
-					Details:            activityInfo.Details,
-					Attempt:            common.Int32Ptr(activityInfo.Attempt),
-					LastFailureReason:  common.StringPtr(activityInfo.LastFailureReason),
-					LastWorkerIdentity: common.StringPtr(activityInfo.LastWorkerIdentity),
-					LastFailureDetails: activityInfo.LastFailureDetails,
-					VersionHistory:     versionHistory,
-				},
-			}, nil
-		},
-	)
-}
-
-func (p *replicatorQueueProcessorImpl) generateHistoryReplicationTask(
-	ctx ctx.Context,
-	task *persistence.ReplicationTaskInfo,
-) (*replicator.ReplicationTask, error) {
-
-	return p.processReplication(
-		ctx,
-		true, // still necessary to send out history replication message if workflow closed
-		task.GetDomainID(),
-		task.GetWorkflowID(),
-		task.GetRunID(),
-		func(mutableState execution.MutableState) (*replicator.ReplicationTask, error) {
-
-			versionHistories := mutableState.GetVersionHistories()
-
-			// TODO when 3+DC migration is done, remove this block of code
-			if versionHistories == nil {
-				domainEntry, err := p.shard.GetDomainCache().GetDomainByID(task.DomainID)
-				if err != nil {
-					return nil, err
-				}
-
-				var targetClusters []string
-				for _, cluster := range domainEntry.GetReplicationConfig().Clusters {
-					targetClusters = append(targetClusters, cluster.ClusterName)
-				}
-
-				replicationTask, newRunID, err := GenerateReplicationTask(
-					targetClusters,
-					task,
-					p.historyV2Mgr,
-					p.metricsClient,
-					nil,
-					common.IntPtr(p.shard.GetShardID()),
-				)
-				if err != nil {
-					return nil, err
-				}
-				if newRunID != "" {
-					isNDCWorkflow, err := p.isNewRunNDCEnabled(ctx, task.DomainID, task.WorkflowID, newRunID)
-					if err != nil {
-						return nil, err
-					}
-					replicationTask.HistoryTaskAttributes.NewRunNDC = common.BoolPtr(isNDCWorkflow)
-				}
-
-				return replicationTask, err
-			}
-
-			// NDC workflow
-			versionHistoryItems, branchToken, err := p.getVersionHistoryItems(
-				mutableState,
-				task.FirstEventID,
-				task.Version,
-			)
-			if err != nil {
-				return nil, err
-			}
-
-			// BranchToken will not set in get dlq replication message request
-			if len(task.BranchToken) == 0 {
-				task.BranchToken = branchToken
-			}
-
-			eventsBlob, err := p.getEventsBlob(
-				task.BranchToken,
-				task.FirstEventID,
-				task.NextEventID,
-			)
-			if err != nil {
-				return nil, err
-			}
-
-			var newRunEventsBlob *shared.DataBlob
-			if len(task.NewRunBranchToken) != 0 {
-				// only get the first batch
-				newRunEventsBlob, err = p.getEventsBlob(
-					task.NewRunBranchToken,
-					common.FirstEventID,
-					common.FirstEventID+1,
-				)
-				if err != nil {
-					return nil, err
-				}
-			}
-
-			replicationTask := &replicator.ReplicationTask{
-				TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeHistoryV2),
-				HistoryTaskV2Attributes: &replicator.HistoryTaskV2Attributes{
-					TaskId:              common.Int64Ptr(task.FirstEventID),
-					DomainId:            common.StringPtr(task.DomainID),
-					WorkflowId:          common.StringPtr(task.WorkflowID),
-					RunId:               common.StringPtr(task.RunID),
-					VersionHistoryItems: versionHistoryItems,
-					Events:              eventsBlob,
-					NewRunEvents:        newRunEventsBlob,
-				},
-			}
-			return replicationTask, nil
-		},
-	)
-}
-
-func (p *replicatorQueueProcessorImpl) generateFailoverMarkerTask(
-	taskInfo *persistence.ReplicationTaskInfo,
-) *replicator.ReplicationTask {
-
-	return &replicator.ReplicationTask{
-		TaskType:     replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeFailoverMarker),
-		SourceTaskId: common.Int64Ptr(taskInfo.GetTaskID()),
-		FailoverMarkerAttributes: &replicator.FailoverMarkerAttributes{
-			DomainID:        common.StringPtr(taskInfo.GetDomainID()),
-			FailoverVersion: common.Int64Ptr(taskInfo.GetVersion()),
-			CreationTime:    common.Int64Ptr(taskInfo.CreationTime),
-		},
-	}
-}
-
-func (p *replicatorQueueProcessorImpl) getEventsBlob(
-	branchToken []byte,
-	firstEventID int64,
-	nextEventID int64,
-) (*shared.DataBlob, error) {
-
-	var eventBatchBlobs []*persistence.DataBlob
-	var pageToken []byte
-	req := &persistence.ReadHistoryBranchRequest{
-		BranchToken:   branchToken,
-		MinEventID:    firstEventID,
-		MaxEventID:    nextEventID,
-		PageSize:      p.shard.GetConfig().ReplicationTaskProcessorReadHistoryBatchSize(),
-		NextPageToken: pageToken,
-		ShardID:       common.IntPtr(p.shard.GetShardID()),
-	}
-
-	for {
-		resp, err := p.historyV2Mgr.ReadRawHistoryBranch(req)
-		if err != nil {
-			return nil, err
-		}
-
-		req.NextPageToken = resp.NextPageToken
-		eventBatchBlobs = append(eventBatchBlobs, resp.HistoryEventBlobs...)
-
-		if len(req.NextPageToken) == 0 {
-			break
-		}
-	}
-
-	if len(eventBatchBlobs) != 1 {
-		return nil, &shared.InternalServiceError{
-			Message: "replicatorQueueProcessor encounter more than 1 NDC raw event batch",
-		}
-	}
-
-	return eventBatchBlobs[0].ToThrift(), nil
-}
-
-func (p *replicatorQueueProcessorImpl) getVersionHistoryItems(
-	mutableState execution.MutableState,
-	eventID int64,
-	version int64,
-) ([]*shared.VersionHistoryItem, []byte, error) {
-
-	versionHistories := mutableState.GetVersionHistories()
-	if versionHistories == nil {
-		return nil, nil, &shared.InternalServiceError{
-			Message: "replicatorQueueProcessor encounter workflow without version histories",
-		}
-	}
-
-	versionHistoryIndex, err := versionHistories.FindFirstVersionHistoryIndexByItem(
-		persistence.NewVersionHistoryItem(
-			eventID,
-			version,
-		),
-	)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	versionHistory, err := versionHistories.GetVersionHistory(versionHistoryIndex)
-	if err != nil {
-		return nil, nil, err
-	}
-	return versionHistory.ToThrift().Items, versionHistory.GetBranchToken(), nil
-}
-
-func (p *replicatorQueueProcessorImpl) processReplication(
-	ctx ctx.Context,
-	processTaskIfClosed bool,
-	domainID string,
-	workflowID string,
-	runID string,
-	action func(execution.MutableState) (*replicator.ReplicationTask, error),
-) (retReplicationTask *replicator.ReplicationTask, retError error) {
-
-	execution := shared.WorkflowExecution{
-		WorkflowId: common.StringPtr(workflowID),
-		RunId:      common.StringPtr(runID),
-	}
-
-	context, release, err := p.executionCache.GetOrCreateWorkflowExecution(ctx, domainID, execution)
-	if err != nil {
-		return nil, err
-	}
-	defer func() { release(retError) }()
-
-	msBuilder, err := context.LoadWorkflowExecution()
-	switch err.(type) {
-	case nil:
-		if !processTaskIfClosed && !msBuilder.IsWorkflowExecutionRunning() {
-			// workflow already finished, no need to process the replication task
-			return nil, nil
-		}
-		return action(msBuilder)
-	case *shared.EntityNotExistsError:
-		return nil, nil
-	default:
-		return nil, err
-	}
-}
-
-func (p *replicatorQueueProcessorImpl) isNewRunNDCEnabled(
-	ctx ctx.Context,
-	domainID string,
-	workflowID string,
-	runID string,
-) (isNDCWorkflow bool, retError error) {
-
-	context, release, err := p.executionCache.GetOrCreateWorkflowExecution(
-		ctx,
-		domainID,
-		shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	)
-	if err != nil {
-		return false, err
-	}
-	defer func() { release(retError) }()
-
-	mutableState, err := context.LoadWorkflowExecution()
-	if err != nil {
-		return false, err
-	}
-	return mutableState.GetVersionHistories() != nil, nil
-}
diff --git a/service/history/replicatorQueueProcessor_mock.go b/service/history/replicatorQueueProcessor_mock.go
deleted file mode 100644
index a3a20ea37..000000000
--- a/service/history/replicatorQueueProcessor_mock.go
+++ /dev/null
@@ -1,117 +0,0 @@
-// Copyright (c) 2017 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-// Code generated by MockGen. DO NOT EDIT.
-// Source: github.com/uber/cadence/service/history (interfaces: ReplicatorQueueProcessor)
-
-// Package history is a generated GoMock package.
-package history
-
-import (
-	"context"
-	reflect "reflect"
-
-	gomock "github.com/golang/mock/gomock"
-
-	replicator "github.com/uber/cadence/.gen/go/replicator"
-)
-
-// MockReplicatorQueueProcessor is a mock of ReplicatorQueueProcessor interface
-type MockReplicatorQueueProcessor struct {
-	ctrl     *gomock.Controller
-	recorder *MockReplicatorQueueProcessorMockRecorder
-}
-
-// MockReplicatorQueueProcessorMockRecorder is the mock recorder for MockReplicatorQueueProcessor
-type MockReplicatorQueueProcessorMockRecorder struct {
-	mock *MockReplicatorQueueProcessor
-}
-
-// NewMockReplicatorQueueProcessor creates a new mock instance
-func NewMockReplicatorQueueProcessor(ctrl *gomock.Controller) *MockReplicatorQueueProcessor {
-	mock := &MockReplicatorQueueProcessor{ctrl: ctrl}
-	mock.recorder = &MockReplicatorQueueProcessorMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use
-func (m *MockReplicatorQueueProcessor) EXPECT() *MockReplicatorQueueProcessorMockRecorder {
-	return m.recorder
-}
-
-// Start mocks base method
-func (m *MockReplicatorQueueProcessor) Start() {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Start")
-}
-
-// Start indicates an expected call of Start
-func (mr *MockReplicatorQueueProcessorMockRecorder) Start() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockReplicatorQueueProcessor)(nil).Start))
-}
-
-// Stop mocks base method
-func (m *MockReplicatorQueueProcessor) Stop() {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Stop")
-}
-
-// Stop indicates an expected call of Stop
-func (mr *MockReplicatorQueueProcessorMockRecorder) Stop() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockReplicatorQueueProcessor)(nil).Stop))
-}
-
-// getTasks mocks base method
-func (m *MockReplicatorQueueProcessor) getTasks(arg0 context.Context, arg1 string, arg2 int64) (*replicator.ReplicationMessages, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "getTasks", arg0, arg1, arg2)
-	ret0, _ := ret[0].(*replicator.ReplicationMessages)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// getTasks mocks base method
-func (m *MockReplicatorQueueProcessor) getTask(arg0 context.Context, arg1 *replicator.ReplicationTaskInfo) (*replicator.ReplicationTask, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "getTask", arg0, arg1)
-	ret0, _ := ret[0].(*replicator.ReplicationTask)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// getTasks indicates an expected call of getTasks
-func (mr *MockReplicatorQueueProcessorMockRecorder) getTasks(arg0 interface{}, arg1 interface{}, arg2 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getTasks", reflect.TypeOf((*MockReplicatorQueueProcessor)(nil).getTasks), arg0, arg1, arg2)
-}
-
-// notifyNewTask mocks base method
-func (m *MockReplicatorQueueProcessor) notifyNewTask() {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "notifyNewTask")
-}
-
-// notifyNewTask indicates an expected call of notifyNewTask
-func (mr *MockReplicatorQueueProcessorMockRecorder) notifyNewTask() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "notifyNewTask", reflect.TypeOf((*MockReplicatorQueueProcessor)(nil).notifyNewTask))
-}
diff --git a/tools/cli/admin.go b/tools/cli/admin.go
index 9a7e82e37..2abd7d6bf 100644
--- a/tools/cli/admin.go
+++ b/tools/cli/admin.go
@@ -409,119 +409,20 @@ func newAdminKafkaCommands() []cli.Command {
 				AdminKafkaParse(c)
 			},
 		},
-		{
-			Name:    "purgeTopic",
-			Aliases: []string{"purge"},
-			Usage:   "purge Kafka topic by consumer group",
-			Flags: []cli.Flag{
-				cli.StringFlag{
-					Name:  FlagCluster,
-					Usage: "Name of the Kafka cluster to publish replicationTasks",
-				},
-				cli.StringFlag{
-					Name:  FlagTopic,
-					Usage: "Topic to publish replication task",
-				},
-				cli.StringFlag{
-					Name:  FlagGroup,
-					Usage: "Group to read DLQ",
-				},
-				cli.StringFlag{
-					Name: FlagHostFile,
-					Usage: "Kafka host config file in format of: " + `
-tls:
-    enabled: false
-    certFile: ""
-    keyFile: ""
-    caFile: ""
-clusters:
-	localKafka:
-		brokers:
-		- 127.0.0.1
-		- 127.0.0.2`,
-				},
-			},
-			Action: func(c *cli.Context) {
-				AdminPurgeTopic(c)
-			},
-		},
-		{
-			Name:    "mergeDLQ",
-			Aliases: []string{"mgdlq"},
-			Usage:   "Merge replication tasks to target topic(from input file or DLQ topic)",
-			Flags: []cli.Flag{
-				cli.StringFlag{
-					Name:  FlagInputFileWithAlias,
-					Usage: "Input file to use to read as JSON of ReplicationTask, separated by line",
-				},
-				cli.StringFlag{
-					Name:  FlagInputTopicWithAlias,
-					Usage: "Input topic to read ReplicationTask",
-				},
-				cli.StringFlag{
-					Name:  FlagInputCluster,
-					Usage: "Name of the Kafka cluster for reading DLQ topic for ReplicationTask",
-				},
-				cli.Int64Flag{
-					Name:  FlagStartOffset,
-					Usage: "Starting offset for reading DLQ topic for ReplicationTask",
-				},
-				cli.StringFlag{
-					Name:  FlagCluster,
-					Usage: "Name of the Kafka cluster to publish replicationTasks",
-				},
-				cli.StringFlag{
-					Name:  FlagTopic,
-					Usage: "Topic to publish replication task",
-				},
-				cli.StringFlag{
-					Name:  FlagGroup,
-					Usage: "Group to read DLQ",
-				},
-				cli.StringFlag{
-					Name: FlagHostFile,
-					Usage: "Kafka host config file in format of: " + `
-tls:
-    enabled: false
-    certFile: ""
-    keyFile: ""
-    caFile: ""
-clusters:
-	localKafka:
-		brokers:
-		- 127.0.0.1
-		- 127.0.0.2`,
-				},
-			},
-			Action: func(c *cli.Context) {
-				AdminMergeDLQ(c)
-			},
-		},
 		{
 			Name:    "rereplicate",
 			Aliases: []string{"rrp"},
 			Usage:   "Rereplicate replication tasks to target topic from history tables",
 			Flags: append(getDBFlags(),
 				cli.StringFlag{
-					Name:  FlagTargetCluster,
-					Usage: "Name of targetCluster to receive the replication task",
+					Name:  FlagSourceCluster,
+					Usage: "Name of source cluster to resend the replication task",
 				},
 				cli.IntFlag{
 					Name:  FlagNumberOfShards,
 					Usage: "NumberOfShards is required to calculate shardID. (see server config for numHistoryShards)",
 				},
-
-				// for multiple workflow
-				cli.StringFlag{
-					Name:  FlagInputFileWithAlias,
-					Usage: "Input file to read multiple workflow line by line. For each line: domainID,workflowID,runID,minEventID,maxEventID (minEventID/maxEventID are optional.)",
-				},
-
 				// for one workflow
-				cli.Int64Flag{
-					Name:  FlagMinEventID,
-					Usage: "MinEventID. Optional, default to all events",
-				},
 				cli.Int64Flag{
 					Name:  FlagMaxEventID,
 					Usage: "MaxEventID Optional, default to all events",
@@ -539,31 +440,8 @@ clusters:
 					Usage: "DomainID",
 				},
 				cli.StringFlag{
-					Name:  FlagStartEventVersion,
-					Usage: "Workflow start event version",
-				},
-				// kafka
-				cli.StringFlag{
-					Name:  FlagCluster,
-					Usage: "Name of the Kafka cluster to publish replicationTasks",
-				},
-				cli.StringFlag{
-					Name:  FlagTopic,
-					Usage: "Topic to publish replication task",
-				},
-				cli.StringFlag{
-					Name: FlagHostFile,
-					Usage: "Kafka host config file in format of: " + `
-tls:
-    enabled: false
-    certFile: ""
-    keyFile: ""
-    caFile: ""
-clusters:
-	localKafka:
-		brokers:
-		- 127.0.0.1
-		- 127.0.0.2`,
+					Name:  FlagEndEventVersion,
+					Usage: "Workflow end event version",
 				}),
 			Action: func(c *cli.Context) {
 				AdminRereplicate(c)
diff --git a/tools/cli/adminKafkaCommands.go b/tools/cli/adminKafkaCommands.go
index 5a05da9f2..8103bc722 100644
--- a/tools/cli/adminKafkaCommands.go
+++ b/tools/cli/adminKafkaCommands.go
@@ -24,27 +24,19 @@ import (
 	"bufio"
 	"bytes"
 	"context"
-	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"regexp"
-	"runtime"
-	"strconv"
-	"strings"
 	"sync/atomic"
 	"time"
 
-	"github.com/Shopify/sarama"
-	cluster "github.com/bsm/sarama-cluster"
 	"github.com/gocql/gocql"
 	"github.com/urfave/cli"
 	"go.uber.org/thriftrw/protocol"
 	"go.uber.org/thriftrw/wire"
-	"gopkg.in/yaml.v2"
 
 	"github.com/uber/cadence/.gen/go/admin"
 	serverAdmin "github.com/uber/cadence/.gen/go/admin/adminserviceclient"
@@ -57,8 +49,6 @@ import (
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/persistence/cassandra"
-	"github.com/uber/cadence/common/service/dynamicconfig"
-	"github.com/uber/cadence/service/history"
 )
 
 type (
@@ -492,132 +482,46 @@ func doRereplicate(
 	domainID string,
 	wid string,
 	rid string,
-	minID int64,
-	maxID int64,
-	startVersion *int64,
-	targets []string,
-	producer messaging.Producer,
+	endEventID int64,
+	endEventVersion int64,
+	sourceCluster string,
 	session *gocql.Session,
 	adminClient serverAdmin.Interface,
 ) {
 
-	if minID <= 0 {
-		minID = 1
-	}
-	if maxID == 0 {
-		maxID = maxRereplicateEventID
-	}
-
-	histV2 := cassandra.NewHistoryV2PersistenceFromSession(session, loggerimpl.NewNopLogger())
-	historyV2Mgr := persistence.NewHistoryV2ManagerImpl(histV2, loggerimpl.NewNopLogger(), dynamicconfig.GetIntPropertyFn(common.DefaultTransactionSizeLimit))
-
 	exeM, _ := cassandra.NewWorkflowExecutionPersistence(shardID, session, loggerimpl.NewNopLogger())
 	exeMgr := persistence.NewExecutionManagerImpl(exeM, loggerimpl.NewNopLogger())
 
-	for {
-		fmt.Printf("Start rereplicate for wid: %v, rid:%v \n", wid, rid)
-		resp, err := exeMgr.GetWorkflowExecution(&persistence.GetWorkflowExecutionRequest{
-			DomainID: domainID,
-			Execution: shared.WorkflowExecution{
-				WorkflowId: common.StringPtr(wid),
-				RunId:      common.StringPtr(rid),
-			},
-		})
-		if err != nil {
-			ErrorAndExit("GetWorkflowExecution error", err)
-		}
-
-		versionHistories := resp.State.VersionHistories
-		if versionHistories != nil {
-			if startVersion == nil {
-				ErrorAndExit("Start event version is require for NDC enabled workflow", nil)
-			}
-			if err := adminClient.ResendReplicationTasks(
-				ctx,
-				&admin.ResendReplicationTasksRequest{
-					DomainID:      common.StringPtr(domainID),
-					WorkflowID:    common.StringPtr(wid),
-					RunID:         common.StringPtr(rid),
-					RemoteCluster: common.StringPtr(targets[0]),
-					StartVersion:  startVersion,
-				},
-			); err != nil {
-				ErrorAndExit("Failed to resend ndc workflow", err)
-			}
-			return
-		}
-
-		currVersion := resp.State.ReplicationState.CurrentVersion
-		repInfo := map[string]*persistence.ReplicationInfo{
-			"": {
-				Version:     currVersion,
-				LastEventID: 0,
-			},
-		}
-
-		exeInfo := resp.State.ExecutionInfo
-		taskTemplate := &persistence.ReplicationTaskInfo{
-			DomainID:            domainID,
-			WorkflowID:          wid,
-			RunID:               rid,
-			Version:             currVersion,
-			LastReplicationInfo: repInfo,
-			BranchToken:         exeInfo.BranchToken,
-		}
-
-		_, historyBatches, err := history.GetAllHistory(historyV2Mgr, nil, true,
-			minID, maxID, exeInfo.BranchToken, common.IntPtr(shardID))
-
-		if err != nil {
-			ErrorAndExit("GetAllHistory error", err)
-		}
-
-		continueAsNew := false
-		var newRunID string
-		for _, batch := range historyBatches {
-
-			events := batch.Events
-			firstEvent := events[0]
-			lastEvent := events[len(events)-1]
-			if lastEvent.GetEventType() == shared.EventTypeWorkflowExecutionContinuedAsNew {
-				continueAsNew = true
-				newRunID = lastEvent.WorkflowExecutionContinuedAsNewEventAttributes.GetNewExecutionRunId()
-				resp, err := exeMgr.GetWorkflowExecution(&persistence.GetWorkflowExecutionRequest{
-					DomainID: domainID,
-					Execution: shared.WorkflowExecution{
-						WorkflowId: common.StringPtr(wid),
-						RunId:      common.StringPtr(newRunID),
-					},
-				})
-				if err != nil {
-					ErrorAndExit("GetWorkflowExecution error", err)
-				}
-				taskTemplate.NewRunBranchToken = resp.State.ExecutionInfo.BranchToken
-			}
-			taskTemplate.Version = firstEvent.GetVersion()
-			taskTemplate.FirstEventID = firstEvent.GetEventId()
-			taskTemplate.NextEventID = lastEvent.GetEventId() + 1
-			task, _, err := history.GenerateReplicationTask(targets, taskTemplate, historyV2Mgr, nil, batch, common.IntPtr(shardID))
-			if err != nil {
-				ErrorAndExit("GenerateReplicationTask error", err)
-			}
-			err = producer.Publish(task)
-			if err != nil {
-				ErrorAndExit("Publish task error", err)
-			}
-			fmt.Printf("publish task successfully firstEventID %v, lastEventID %v \n", firstEvent.GetEventId(), lastEvent.GetEventId())
-		}
-
-		fmt.Printf("Done rereplicate for wid: %v, rid:%v \n", wid, rid)
-		runtime.GC()
-		if continueAsNew {
-			rid = newRunID
-			minID = 1
-			maxID = maxRereplicateEventID
-		} else {
-			break
-		}
-	}
+	fmt.Printf("Start rereplicate for wid: %v, rid:%v \n", wid, rid)
+	resp, err := exeMgr.GetWorkflowExecution(&persistence.GetWorkflowExecutionRequest{
+		DomainID: domainID,
+		Execution: shared.WorkflowExecution{
+			WorkflowId: common.StringPtr(wid),
+			RunId:      common.StringPtr(rid),
+		},
+	})
+	if err != nil {
+		ErrorAndExit("GetWorkflowExecution error", err)
+	}
+
+	versionHistories := resp.State.VersionHistories
+	if versionHistories == nil {
+		ErrorAndExit("The workflow is not a NDC workflow", nil)
+	}
+	if err := adminClient.ResendReplicationTasks(
+		ctx,
+		&admin.ResendReplicationTasksRequest{
+			DomainID:      common.StringPtr(domainID),
+			WorkflowID:    common.StringPtr(wid),
+			RunID:         common.StringPtr(rid),
+			RemoteCluster: common.StringPtr(sourceCluster),
+			EndEventID:    common.Int64Ptr(endEventID + 1),
+			EndVersion:    common.Int64Ptr(endEventVersion),
+		},
+	); err != nil {
+		ErrorAndExit("Failed to resend ndc workflow", err)
+	}
+	fmt.Printf("Done rereplicate for wid: %v, rid:%v \n", wid, rid)
 }
 
 // AdminRereplicate parses will re-publish replication tasks to topic
@@ -627,19 +531,22 @@ func AdminRereplicate(c *cli.Context) {
 		ErrorAndExit("numberOfShards is must be > 0", nil)
 		return
 	}
-	target := getRequiredOption(c, FlagTargetCluster)
-	targets := []string{target}
+	sourceCluster := getRequiredOption(c, FlagSourceCluster)
+	if !c.IsSet(FlagMaxEventID) {
+		ErrorAndExit("End event ID is not defined", nil)
+	}
+	if !c.IsSet(FlagEndEventVersion) {
+		ErrorAndExit("End event version is not defined", nil)
+	}
 
 	session := connectToCassandra(c)
 	adminClient := cFactory.ServerAdminClient(c)
-	var startVersion *int64
-	var producer messaging.Producer
-	if c.IsSet(FlagStartEventVersion) {
-		startVersion = common.Int64Ptr(c.Int64(FlagStartEventVersion))
-	} else {
-		producer = newKafkaProducer(c)
-	}
-
+	endEventID := c.Int64(FlagMaxEventID)
+	endVersion := c.Int64(FlagEndEventVersion)
+	domainID := getRequiredOption(c, FlagDomainID)
+	wid := getRequiredOption(c, FlagWorkflowID)
+	rid := getRequiredOption(c, FlagRunID)
+	shardID := common.WorkflowIDToHistoryShard(wid, numberOfShards)
 	contextTimeout := defaultResendContextTimeout
 	if c.GlobalIsSet(FlagContextTimeout) {
 		contextTimeout = time.Duration(c.GlobalInt(FlagContextTimeout)) * time.Second
@@ -647,318 +554,18 @@ func AdminRereplicate(c *cli.Context) {
 	ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 	defer cancel()
 
-	if c.IsSet(FlagInputFile) {
-		inFile := c.String(FlagInputFile)
-		// This code is executed from the CLI. All user input is from a CLI user.
-		// parse domainID,workflowID,runID,minEventID,maxEventID
-		// #nosec
-		file, err := os.Open(inFile)
-		if err != nil {
-			ErrorAndExit("Open failed", err)
-		}
-		defer file.Close()
-
-		scanner := bufio.NewScanner(file)
-		idx := 0
-		for scanner.Scan() {
-			idx++
-			line := strings.TrimSpace(scanner.Text())
-			if len(line) == 0 {
-				fmt.Printf("line %v is empty, skipped\n", idx)
-				continue
-			}
-			cols := strings.Split(line, ",")
-			if len(cols) < 3 {
-				ErrorAndExit("Split failed", fmt.Errorf("line %v has less than 3 cols separated by comma, only %v ", idx, len(cols)))
-			}
-			fmt.Printf("Start processing line %v ...\n", idx)
-			domainID := strings.TrimSpace(cols[0])
-			wid := strings.TrimSpace(cols[1])
-			rid := strings.TrimSpace(cols[2])
-			var minID, maxID int64
-			if len(cols) >= 4 {
-				i, err := strconv.Atoi(strings.TrimSpace(cols[3]))
-				if err != nil {
-					ErrorAndExit(fmt.Sprintf("Atoi failed at lne %v", idx), err)
-				}
-				minID = int64(i)
-			}
-			if len(cols) >= 5 {
-				i, err := strconv.Atoi(strings.TrimSpace(cols[4]))
-				if err != nil {
-					ErrorAndExit(fmt.Sprintf("Atoi failed at lne %v", idx), err)
-				}
-				maxID = int64(i)
-			}
-
-			shardID := common.WorkflowIDToHistoryShard(wid, numberOfShards)
-			doRereplicate(ctx, shardID, domainID, wid, rid, minID, maxID, startVersion, targets, producer, session, adminClient)
-			fmt.Printf("Done processing line %v ...\n", idx)
-		}
-		if err := scanner.Err(); err != nil {
-			ErrorAndExit("scanner failed", err)
-		}
-	} else {
-		domainID := getRequiredOption(c, FlagDomainID)
-		wid := getRequiredOption(c, FlagWorkflowID)
-		rid := getRequiredOption(c, FlagRunID)
-		minID := c.Int64(FlagMinEventID)
-		maxID := c.Int64(FlagMaxEventID)
-		shardID := common.WorkflowIDToHistoryShard(wid, numberOfShards)
-
-		doRereplicate(ctx, shardID, domainID, wid, rid, minID, maxID, startVersion, targets, producer, session, adminClient)
-	}
-}
-
-func newKafkaProducer(c *cli.Context) messaging.Producer {
-	hostFile := getRequiredOption(c, FlagHostFile)
-	destCluster := getRequiredOption(c, FlagCluster)
-	destTopic := getRequiredOption(c, FlagTopic)
-
-	// initialize kafka producer
-	destBrokers, tlsConfig, err := loadBrokerConfig(hostFile, destCluster)
-	if err != nil {
-		ErrorAndExit("", err)
-	}
-
-	config := sarama.NewConfig()
-	config.Producer.RequiredAcks = sarama.WaitForAll
-	config.Producer.Return.Successes = true
-	if tlsConfig != nil {
-		config.Net.TLS.Config = tlsConfig
-		config.Net.TLS.Enable = true
-	}
-	sproducer, err := sarama.NewSyncProducer(destBrokers, config)
-	if err != nil {
-		ErrorAndExit("", err)
-	}
-	logger := loggerimpl.NewNopLogger()
-
-	producer := messaging.NewKafkaProducer(destTopic, sproducer, logger)
-	return producer
-}
-
-// AdminPurgeTopic is used to purge kafka topic
-func AdminPurgeTopic(c *cli.Context) {
-	hostFile := getRequiredOption(c, FlagHostFile)
-	topic := getRequiredOption(c, FlagTopic)
-	cluster := getRequiredOption(c, FlagCluster)
-	group := getRequiredOption(c, FlagGroup)
-	brokers, tlsConfig, err := loadBrokerConfig(hostFile, cluster)
-
-	consumer := createConsumerAndWaitForReady(brokers, tlsConfig, group, topic)
-
-	highWaterMarks, ok := consumer.HighWaterMarks()[topic]
-	if !ok {
-		ErrorAndExit("", fmt.Errorf("cannot find high watermark"))
-	}
-	fmt.Printf("Topic high watermark %v.\n", highWaterMarks)
-	for partition, hi := range highWaterMarks {
-		consumer.MarkPartitionOffset(topic, partition, hi-1, "")
-		fmt.Printf("set partition offset %v:%v \n", partition, hi)
-	}
-	err = consumer.CommitOffsets()
-	if err != nil {
-		ErrorAndExit("fail to commit offset", err)
-	}
-
-	consumer = createConsumerAndWaitForReady(brokers, tlsConfig, group, topic)
-	msg, ok := <-consumer.Messages()
-	if !ok {
-		fmt.Println("consumer channel is closed")
-	}
-	fmt.Printf("current offset sample: %v: %v \n", msg.Partition, msg.Offset)
-}
-
-// AdminMergeDLQ publish replication tasks from DLQ or JSON file
-func AdminMergeDLQ(c *cli.Context) {
-	hostFile := getRequiredOption(c, FlagHostFile)
-	producer := newKafkaProducer(c)
-
-	var err error
-	var inFile string
-	var tasks []*replicator.ReplicationTask
-	if c.IsSet(FlagInputFile) && (c.IsSet(FlagInputCluster) || c.IsSet(FlagInputTopic) || c.IsSet(FlagStartOffset)) {
-		ErrorAndExit("", fmt.Errorf("ONLY Either from JSON file or from DLQ topic"))
-	}
-
-	if c.IsSet(FlagInputFile) {
-		inFile = c.String(FlagInputFile)
-		// parse json input as replicaiton tasks
-		tasks, err = parseReplicationTask(inFile)
-		if err != nil {
-			ErrorAndExit("", err)
-		}
-		// publish to topic
-		for idx, t := range tasks {
-			err := producer.Publish(t)
-			if err != nil {
-				fmt.Printf("cannot publish task %v to topic \n", idx)
-				ErrorAndExit("", err)
-			} else {
-				fmt.Printf("replication task sent: %v firstID %v, nextID %v \n", idx, t.HistoryTaskAttributes.GetFirstEventId(), t.HistoryTaskAttributes.GetNextEventId())
-			}
-		}
-	} else {
-		fromTopic := getRequiredOption(c, FlagInputTopic)
-		fromCluster := getRequiredOption(c, FlagInputCluster)
-		startOffset := c.Int64(FlagStartOffset)
-		group := getRequiredOption(c, FlagGroup)
-
-		fromBrokers, tlsConfig, err := loadBrokerConfig(hostFile, fromCluster)
-		if err != nil {
-			ErrorAndExit("", err)
-		}
-
-		consumer := createConsumerAndWaitForReady(fromBrokers, tlsConfig, group, fromTopic)
-
-		highWaterMarks, ok := consumer.HighWaterMarks()[fromTopic]
-		if !ok {
-			ErrorAndExit("", fmt.Errorf("cannot find high watermark"))
-		}
-		fmt.Printf("Topic high watermark %v.\n", highWaterMarks)
-		for partition := range highWaterMarks {
-			consumer.MarkPartitionOffset(fromTopic, partition, startOffset, "")
-			fmt.Printf("reset offset %v:%v \n", partition, startOffset)
-		}
-		err = consumer.CommitOffsets()
-		if err != nil {
-			ErrorAndExit("fail to commit offset", err)
-		}
-		// create consumer again to make sure MarkPartitionOffset works
-		consumer = createConsumerAndWaitForReady(fromBrokers, tlsConfig, group, fromTopic)
-
-		for {
-			select {
-			case msg, ok := <-consumer.Messages():
-				if !ok {
-					return
-				}
-				if msg.Offset < startOffset {
-					fmt.Printf("Wrong Message [%v],[%v] \n", msg.Partition, msg.Offset)
-					ErrorAndExit("", fmt.Errorf("offset is not correct"))
-					continue
-				} else {
-					var task replicator.ReplicationTask
-					err := decode(msg.Value, &task)
-					if err != nil {
-						ErrorAndExit("failed to deserialize message due to error", err)
-					}
-
-					err = producer.Publish(&task)
-
-					if err != nil {
-						fmt.Printf("[Error] Message [%v],[%v] failed: %v\n", msg.Partition, msg.Offset, err)
-					} else {
-						fmt.Printf("Message [%v],[%v] succeeded\n", msg.Partition, msg.Offset)
-					}
-				}
-				consumer.MarkOffset(msg, "")
-			case <-time.After(time.Second * 5):
-				fmt.Println("heartbeat: waiting for more messages, Ctrl+C to stop any time...")
-			}
-		}
-	}
-}
-
-func createConsumerAndWaitForReady(brokers []string, tlsConfig *tls.Config, group, fromTopic string) *cluster.Consumer {
-	config := cluster.NewConfig()
-	config.Consumer.Return.Errors = true
-	config.Consumer.Offsets.Initial = sarama.OffsetOldest
-	if tlsConfig != nil {
-		config.Net.TLS.Enable = true
-		config.Net.TLS.Config = tlsConfig
-	}
-
-	config.Group.Return.Notifications = true
-
-	client, err := cluster.NewClient(brokers, config)
-	if err != nil {
-		ErrorAndExit("", err)
-	}
-
-	consumer, err := cluster.NewConsumerFromClient(client, group, []string{fromTopic})
-	if err != nil {
-		ErrorAndExit("", err)
-	}
-
-	for ntf := range consumer.Notifications() {
-		time.Sleep(time.Second)
-		if partitions := ntf.Current[fromTopic]; len(partitions) > 0 && ntf.Type == cluster.RebalanceOK {
-			break
-		}
-		fmt.Println("Waiting for consumer ready...")
-	}
-	return consumer
-}
-
-func parseReplicationTask(in string) (tasks []*replicator.ReplicationTask, err error) {
-	// This code is executed from the CLI. All user input is from a CLI user.
-	// #nosec
-	file, err := os.Open(in)
-	if err != nil {
-		return nil, err
-	}
-	defer file.Close()
-
-	reader := bufio.NewReaderSize(file, 1024*1024*3)
-	idx := 0
-	for {
-		idx++
-		line, _, err := reader.ReadLine()
-		if err != nil {
-			if err != io.EOF {
-				ErrorAndExit("Failed to read line", err)
-			}
-			break
-		}
-
-		if len(line) == 0 {
-			fmt.Printf("line %v is empty, skipped\n", idx)
-			continue
-		}
-
-		t := &replicator.ReplicationTask{}
-		err = json.Unmarshal(line, t)
-		if err != nil {
-			fmt.Printf("line %v cannot be deserialized to replicaiton task: %v.\n", idx, string(line))
-			return nil, err
-		}
-		tasks = append(tasks, t)
-	}
-
-	return tasks, nil
-}
-
-func loadBrokerConfig(hostFile string, cluster string) ([]string, *tls.Config, error) {
-	// This code is executed from the CLI and is only used to load config files
-	// #nosec
-	contents, err := ioutil.ReadFile(hostFile)
-	if err != nil {
-		return nil, nil, fmt.Errorf("failed to load kafka cluster info from %v., error: %v", hostFile, err)
-	}
-	clustersConfig := ClustersConfig{}
-	if err := yaml.Unmarshal(contents, &clustersConfig); err != nil {
-		return nil, nil, err
-	}
-	if len(clustersConfig.Clusters) != 0 {
-		config, ok := clustersConfig.Clusters[cluster]
-		if ok {
-			brs := config.Brokers
-			for i, b := range brs {
-				if !strings.Contains(b, ":") {
-					b += ":9092"
-					brs[i] = b
-				}
-			}
-			tlsConfig, err := messaging.CreateTLSConfig(clustersConfig.TLS)
-			if err != nil {
-				return nil, nil, fmt.Errorf(fmt.Sprintf("Error creating Kafka TLS config %v", err))
-			}
-			return brs, tlsConfig, nil
-		}
-	}
-	return nil, nil, fmt.Errorf("failed to load broker for cluster %v", cluster)
+	doRereplicate(
+		ctx,
+		shardID,
+		domainID,
+		wid,
+		rid,
+		endEventID,
+		endVersion,
+		sourceCluster,
+		session,
+		adminClient,
+	)
 }
 
 func decodeReplicationTask(
diff --git a/tools/cli/domainUtils.go b/tools/cli/domainUtils.go
index 03fb4de2d..22f6d81ce 100644
--- a/tools/cli/domainUtils.go
+++ b/tools/cli/domainUtils.go
@@ -324,7 +324,6 @@ func initializeClusterMetadata(
 		clusterMetadata.MasterClusterName,
 		clusterMetadata.CurrentClusterName,
 		clusterMetadata.ClusterInformation,
-		clusterMetadata.ReplicationConsumer,
 	)
 }
 
diff --git a/tools/cli/flags.go b/tools/cli/flags.go
index 7b8a28c5c..67791a5f1 100644
--- a/tools/cli/flags.go
+++ b/tools/cli/flags.go
@@ -49,7 +49,7 @@ const (
 	FlagSourceCluster                     = "source_cluster"
 	FlagMinEventID                        = "min_event_id"
 	FlagMaxEventID                        = "max_event_id"
-	FlagStartEventVersion                 = "start_event_version"
+	FlagEndEventVersion                   = "end_event_version"
 	FlagTaskList                          = "tasklist"
 	FlagTaskListWithAlias                 = FlagTaskList + ", tl"
 	FlagTaskListType                      = "tasklisttype"


Test patch:
diff --git a/common/cache/domainCache_test.go b/common/cache/domainCache_test.go
index 9f77ec71b..5f0fe2ea5 100644
--- a/common/cache/domainCache_test.go
+++ b/common/cache/domainCache_test.go
@@ -39,7 +39,6 @@ import (
 	"github.com/uber/cadence/common/metrics"
 	"github.com/uber/cadence/common/mocks"
 	"github.com/uber/cadence/common/persistence"
-	"github.com/uber/cadence/common/service/config"
 	"github.com/uber/cadence/common/service/dynamicconfig"
 )
 
@@ -664,9 +663,6 @@ func Test_DomainCacheEntry_GetDomainNotActiveErr(t *testing.T) {
 		cluster.TestCurrentClusterName,
 		cluster.TestCurrentClusterName,
 		cluster.TestAllClusterInfo,
-		&config.ReplicationConsumerConfig{
-			Type: config.ReplicationConsumerTypeRPC,
-		},
 	)
 	domainEntry := NewGlobalDomainCacheEntryForTest(
 		&persistence.DomainInfo{Name: "test-domain"},
diff --git a/common/cluster/metadataTest.go b/common/cluster/metadataTest.go
index c3dc457f7..03ee9cd68 100644
--- a/common/cluster/metadataTest.go
+++ b/common/cluster/metadataTest.go
@@ -91,9 +91,6 @@ func GetTestClusterMetadata(enableGlobalDomain bool, isMasterCluster bool) Metad
 			masterClusterName,
 			TestCurrentClusterName,
 			TestAllClusterInfo,
-			&config.ReplicationConsumerConfig{
-				Type: config.ReplicationConsumerTypeRPC,
-			},
 		)
 	}
 
@@ -104,6 +101,5 @@ func GetTestClusterMetadata(enableGlobalDomain bool, isMasterCluster bool) Metad
 		TestCurrentClusterName,
 		TestCurrentClusterName,
 		TestSingleDCClusterInfo,
-		nil,
 	)
 }
diff --git a/common/persistence/persistence-tests/executionManagerTest.go b/common/persistence/persistence-tests/executionManagerTest.go
index 0b374e469..bb5e41c73 100644
--- a/common/persistence/persistence-tests/executionManagerTest.go
+++ b/common/persistence/persistence-tests/executionManagerTest.go
@@ -785,9 +785,9 @@ func (s *ExecutionManagerSuite) TestUpsertWorkflowActivity() {
 			Checksum:       csum,
 			UpsertActivityInfos: []*p.ActivityInfo{
 				&p.ActivityInfo{
-					Version: 0,
+					Version:    0,
 					ScheduleID: 100,
-					TaskList: "test-activity-tasktlist-1",
+					TaskList:   "test-activity-tasktlist-1",
 				},
 			},
 		},
@@ -810,9 +810,9 @@ func (s *ExecutionManagerSuite) TestUpsertWorkflowActivity() {
 			Checksum:       csum,
 			UpsertActivityInfos: []*p.ActivityInfo{
 				&p.ActivityInfo{
-					Version: 0,
+					Version:    0,
 					ScheduleID: 100,
-					TaskList: "test-activity-tasktlist-2",
+					TaskList:   "test-activity-tasktlist-2",
 				},
 			},
 		},
diff --git a/host/testcluster.go b/host/testcluster.go
index 099f672b7..65f5e71dc 100644
--- a/host/testcluster.go
+++ b/host/testcluster.go
@@ -114,7 +114,6 @@ func NewCluster(options *TestClusterConfig, logger log.Logger) (*TestCluster, er
 			options.ClusterMetadata.MasterClusterName,
 			options.ClusterMetadata.CurrentClusterName,
 			options.ClusterMetadata.ClusterInformation,
-			options.ClusterMetadata.ReplicationConsumer,
 		)
 	}
 
diff --git a/service/history/conflictResolver_test.go b/service/history/conflictResolver_test.go
index ddb5a788f..3fd0a9e77 100644
--- a/service/history/conflictResolver_test.go
+++ b/service/history/conflictResolver_test.go
@@ -54,14 +54,13 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockEventsCache          *events.MockCache
-		mockDomainCache          *cache.MockDomainCache
-		mockClusterMetadata      *cluster.MockMetadata
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockTxProcessor     *queue.MockProcessor
+		mockTimerProcessor  *queue.MockProcessor
+		mockEventsCache     *events.MockCache
+		mockDomainCache     *cache.MockDomainCache
+		mockClusterMetadata *cluster.MockMetadata
 
 		logger           log.Logger
 		mockExecutionMgr *mocks.ExecutionManager
@@ -90,10 +89,8 @@ func (s *conflictResolverSuite) SetupTest() {
 	s.controller = gomock.NewController(s.T())
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -120,7 +117,6 @@ func (s *conflictResolverSuite) SetupTest() {
 		clusterMetadata:      s.mockClusterMetadata,
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 	}
 	s.mockShard.SetEngine(h)
diff --git a/service/history/historyEngine2_test.go b/service/history/historyEngine2_test.go
index 23e9add62..a4001f88e 100644
--- a/service/history/historyEngine2_test.go
+++ b/service/history/historyEngine2_test.go
@@ -61,14 +61,13 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockEventsCache          *events.MockCache
-		mockDomainCache          *cache.MockDomainCache
-		mockClusterMetadata      *cluster.MockMetadata
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockTxProcessor     *queue.MockProcessor
+		mockTimerProcessor  *queue.MockProcessor
+		mockEventsCache     *events.MockCache
+		mockDomainCache     *cache.MockDomainCache
+		mockClusterMetadata *cluster.MockMetadata
 
 		historyEngine    *historyEngineImpl
 		mockExecutionMgr *mocks.ExecutionManager
@@ -98,10 +97,8 @@ func (s *engine2Suite) SetupTest() {
 
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -145,7 +142,6 @@ func (s *engine2Suite) SetupTest() {
 		timeSource:           s.mockShard.GetTimeSource(),
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 	}
 	s.mockShard.SetEngine(h)
diff --git a/service/history/historyEngine3_eventsv2_test.go b/service/history/historyEngine3_eventsv2_test.go
index 3e6eb6288..b4780de8a 100644
--- a/service/history/historyEngine3_eventsv2_test.go
+++ b/service/history/historyEngine3_eventsv2_test.go
@@ -57,14 +57,13 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockEventsCache          *events.MockCache
-		mockDomainCache          *cache.MockDomainCache
-		mockClusterMetadata      *cluster.MockMetadata
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockTxProcessor     *queue.MockProcessor
+		mockTimerProcessor  *queue.MockProcessor
+		mockEventsCache     *events.MockCache
+		mockDomainCache     *cache.MockDomainCache
+		mockClusterMetadata *cluster.MockMetadata
 
 		historyEngine    *historyEngineImpl
 		mockExecutionMgr *mocks.ExecutionManager
@@ -93,10 +92,8 @@ func (s *engine3Suite) SetupTest() {
 	s.controller = gomock.NewController(s.T())
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -136,7 +133,6 @@ func (s *engine3Suite) SetupTest() {
 		timeSource:           s.mockShard.GetTimeSource(),
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 	}
 	s.mockShard.SetEngine(h)
diff --git a/service/history/historyEngine_reset_test.go b/service/history/historyEngine_reset_test.go
index 12e3fb130..5c8ec3172 100644
--- a/service/history/historyEngine_reset_test.go
+++ b/service/history/historyEngine_reset_test.go
@@ -57,14 +57,13 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockEventsCache          *events.MockCache
-		mockDomainCache          *cache.MockDomainCache
-		mockClusterMetadata      *cluster.MockMetadata
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockTxProcessor     *queue.MockProcessor
+		mockTimerProcessor  *queue.MockProcessor
+		mockEventsCache     *events.MockCache
+		mockDomainCache     *cache.MockDomainCache
+		mockClusterMetadata *cluster.MockMetadata
 
 		mockExecutionMgr *mocks.ExecutionManager
 		mockHistoryV2Mgr *mocks.HistoryV2Manager
@@ -99,10 +98,8 @@ func (s *resetorSuite) SetupTest() {
 	s.controller = gomock.NewController(s.T())
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -140,7 +137,6 @@ func (s *resetorSuite) SetupTest() {
 		config:               s.config,
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), s.mockShard.GetMetricsClient(), func(string) int { return 0 }),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 	}
 	s.mockShard.SetEngine(h)
diff --git a/service/history/historyEngine_test.go b/service/history/historyEngine_test.go
index 46a4991e0..00070e3a9 100644
--- a/service/history/historyEngine_test.go
+++ b/service/history/historyEngine_test.go
@@ -67,17 +67,16 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockDomainCache          *cache.MockDomainCache
-		mockMatchingClient       *matchingservicetest.MockClient
-		mockHistoryClient        *historyservicetest.MockClient
-		mockClusterMetadata      *cluster.MockMetadata
-		mockEventsReapplier      *ndc.MockEventsReapplier
-		mockWorkflowResetter     *reset.MockWorkflowResetter
+		controller           *gomock.Controller
+		mockShard            *shard.TestContext
+		mockTxProcessor      *queue.MockProcessor
+		mockTimerProcessor   *queue.MockProcessor
+		mockDomainCache      *cache.MockDomainCache
+		mockMatchingClient   *matchingservicetest.MockClient
+		mockHistoryClient    *historyservicetest.MockClient
+		mockClusterMetadata  *cluster.MockMetadata
+		mockEventsReapplier  *ndc.MockEventsReapplier
+		mockWorkflowResetter *reset.MockWorkflowResetter
 
 		mockHistoryEngine *historyEngineImpl
 		mockExecutionMgr  *mocks.ExecutionManager
@@ -107,12 +106,10 @@ func (s *engineSuite) SetupTest() {
 	s.controller = gomock.NewController(s.T())
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockEventsReapplier = ndc.NewMockEventsReapplier(s.controller)
 	s.mockWorkflowResetter = reset.NewMockWorkflowResetter(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -166,7 +163,6 @@ func (s *engineSuite) SetupTest() {
 		historyEventNotifier: historyEventNotifier,
 		config:               config.NewForTest(),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 		clientChecker:        cc.NewVersionChecker(),
 		eventsReapplier:      s.mockEventsReapplier,
diff --git a/service/history/historyReplicator_test.go b/service/history/historyReplicator_test.go
index ae75d4bde..b3a46cf3f 100644
--- a/service/history/historyReplicator_test.go
+++ b/service/history/historyReplicator_test.go
@@ -65,15 +65,14 @@ type (
 		suite.Suite
 		*require.Assertions
 
-		controller               *gomock.Controller
-		mockShard                *shard.TestContext
-		mockWorkflowResetor      *reset.MockWorkflowResetor
-		mockTxProcessor          *queue.MockProcessor
-		mockTimerProcessor       *queue.MockProcessor
-		mockReplicationProcessor *MockReplicatorQueueProcessor
-		mockStateBuilder         *execution.MockStateBuilder
-		mockDomainCache          *cache.MockDomainCache
-		mockClusterMetadata      *cluster.MockMetadata
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockWorkflowResetor *reset.MockWorkflowResetor
+		mockTxProcessor     *queue.MockProcessor
+		mockTimerProcessor  *queue.MockProcessor
+		mockStateBuilder    *execution.MockStateBuilder
+		mockDomainCache     *cache.MockDomainCache
+		mockClusterMetadata *cluster.MockMetadata
 
 		logger           log.Logger
 		mockExecutionMgr *mocks.ExecutionManager
@@ -104,11 +103,9 @@ func (s *historyReplicatorSuite) SetupTest() {
 	s.mockWorkflowResetor = reset.NewMockWorkflowResetor(s.controller)
 	s.mockTxProcessor = queue.NewMockProcessor(s.controller)
 	s.mockTimerProcessor = queue.NewMockProcessor(s.controller)
-	s.mockReplicationProcessor = NewMockReplicatorQueueProcessor(s.controller)
 	s.mockStateBuilder = execution.NewMockStateBuilder(s.controller)
 	s.mockTxProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
 	s.mockTimerProcessor.EXPECT().NotifyNewTask(gomock.Any(), gomock.Any()).AnyTimes()
-	s.mockReplicationProcessor.EXPECT().notifyNewTask().AnyTimes()
 
 	s.mockShard = shard.NewTestContext(
 		s.controller,
@@ -144,7 +141,6 @@ func (s *historyReplicatorSuite) SetupTest() {
 		timeSource:           s.mockShard.GetTimeSource(),
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
 		txProcessor:          s.mockTxProcessor,
-		replicatorProcessor:  s.mockReplicationProcessor,
 		timerProcessor:       s.mockTimerProcessor,
 	}
 	s.mockShard.SetEngine(engine)
diff --git a/service/history/replication/task_ack_manager_test.go b/service/history/replication/task_ack_manager_test.go
new file mode 100644
index 000000000..f910b00bf
--- /dev/null
+++ b/service/history/replication/task_ack_manager_test.go
@@ -0,0 +1,951 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package replication
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/pborman/uuid"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/uber/cadence/.gen/go/replicator"
+	workflow "github.com/uber/cadence/.gen/go/shared"
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/cache"
+	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/mocks"
+	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/service/history/config"
+	"github.com/uber/cadence/service/history/execution"
+	"github.com/uber/cadence/service/history/shard"
+)
+
+type (
+	taskAckManagerSuite struct {
+		suite.Suite
+		*require.Assertions
+
+		controller          *gomock.Controller
+		mockShard           *shard.TestContext
+		mockDomainCache     *cache.MockDomainCache
+		mockMutableState    *execution.MockMutableState
+		mockClusterMetadata *cluster.MockMetadata
+
+		mockExecutionMgr *mocks.ExecutionManager
+		mockHistoryMgr   *mocks.HistoryV2Manager
+
+		logger log.Logger
+
+		ackManager *taskAckManagerImpl
+	}
+)
+
+func TestTaskAckManagerSuite(t *testing.T) {
+	s := new(taskAckManagerSuite)
+	suite.Run(t, s)
+}
+
+func (s *taskAckManagerSuite) SetupSuite() {
+
+}
+
+func (s *taskAckManagerSuite) TearDownSuite() {
+
+}
+
+func (s *taskAckManagerSuite) SetupTest() {
+	s.Assertions = require.New(s.T())
+
+	s.controller = gomock.NewController(s.T())
+	s.mockMutableState = execution.NewMockMutableState(s.controller)
+
+	s.mockShard = shard.NewTestContext(
+		s.controller,
+		&persistence.ShardInfo{
+			ShardID:                 0,
+			RangeID:                 1,
+			TransferAckLevel:        0,
+			ClusterReplicationLevel: make(map[string]int64),
+		},
+		config.NewForTest(),
+	)
+
+	s.mockDomainCache = s.mockShard.Resource.DomainCache
+	s.mockExecutionMgr = s.mockShard.Resource.ExecutionMgr
+	s.mockHistoryMgr = s.mockShard.Resource.HistoryMgr
+	s.mockClusterMetadata = s.mockShard.Resource.ClusterMetadata
+	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	s.mockClusterMetadata.EXPECT().IsGlobalDomainEnabled().Return(true).AnyTimes()
+
+	s.logger = s.mockShard.GetLogger()
+	executionCache := execution.NewCache(s.mockShard)
+
+	s.ackManager = NewTaskAckManager(
+		s.mockShard,
+		executionCache,
+	).(*taskAckManagerImpl)
+}
+
+func (s *taskAckManagerSuite) TearDownTest() {
+	s.controller.Finish()
+	s.mockShard.Finish(s.T())
+}
+
+func (s *taskAckManagerSuite) TestConvertLastReplicationInfo() {
+	info := map[string]*persistence.ReplicationInfo{
+		"test": {
+			Version:     0,
+			LastEventID: 0,
+		}}
+	replicationInfo := convertLastReplicationInfo(info)
+	s.NotNil(replicationInfo["test"])
+	s.Equal(info["test"].Version, replicationInfo["test"].GetVersion())
+	s.Equal(info["test"].LastEventID, replicationInfo["test"].GetLastEventId())
+}
+
+func (s *taskAckManagerSuite) TestGetPaginationFunc() {
+	firstEventID := int64(0)
+	nextEventID := int64(1)
+	var branchToken []byte
+	shardID := 0
+	historyCount := 0
+	pagingFunc := s.ackManager.getPaginationFunc(firstEventID, nextEventID, branchToken, shardID, &historyCount)
+
+	pageToken := []byte{1}
+	event := &workflow.HistoryEvent{
+		EventId: common.Int64Ptr(1),
+	}
+	s.mockHistoryMgr.On("ReadHistoryBranch", mock.Anything).Return(&persistence.ReadHistoryBranchResponse{
+		HistoryEvents:    []*workflow.HistoryEvent{event},
+		NextPageToken:    pageToken,
+		Size:             1,
+		LastFirstEventID: 1,
+	}, nil)
+	events, token, err := pagingFunc(nil)
+	s.NoError(err)
+	s.Equal(pageToken, token)
+	s.Len(events, 1)
+	s.Equal(events[0].(*workflow.HistoryEvent), event)
+	s.Equal(historyCount, 1)
+}
+
+func (s *taskAckManagerSuite) TestGetAllHistory_OK() {
+	firstEventID := int64(0)
+	nextEventID := int64(1)
+	var branchToken []byte
+	event := &workflow.HistoryEvent{
+		EventId: common.Int64Ptr(1),
+	}
+
+	s.mockHistoryMgr.On("ReadHistoryBranch", mock.Anything).Return(&persistence.ReadHistoryBranchResponse{
+		HistoryEvents:    []*workflow.HistoryEvent{event},
+		NextPageToken:    nil,
+		Size:             1,
+		LastFirstEventID: 1,
+	}, nil)
+
+	history, err := s.ackManager.getAllHistory(firstEventID, nextEventID, branchToken)
+	s.NoError(err)
+	s.Len(history.GetEvents(), 1)
+	s.Equal(event, history.GetEvents()[0])
+}
+
+func (s *taskAckManagerSuite) TestGetAllHistory_Error() {
+	firstEventID := int64(0)
+	nextEventID := int64(1)
+	var branchToken []byte
+	s.mockHistoryMgr.On("ReadHistoryBranch", mock.Anything).Return(nil, errors.New("test"))
+
+	history, err := s.ackManager.getAllHistory(firstEventID, nextEventID, branchToken)
+	s.Error(err)
+	s.Nil(history)
+}
+
+func (s *taskAckManagerSuite) TestReadTasksWithBatchSize_OK() {
+	task := &persistence.ReplicationTaskInfo{
+		DomainID: uuid.New(),
+	}
+	s.mockExecutionMgr.On("GetReplicationTasks", mock.Anything).Return(&persistence.GetReplicationTasksResponse{
+		Tasks:         []*persistence.ReplicationTaskInfo{task},
+		NextPageToken: []byte{1},
+	}, nil)
+
+	taskInfo, hasMore, err := s.ackManager.readTasksWithBatchSize(0, 1)
+	s.NoError(err)
+	s.True(hasMore)
+	s.Len(taskInfo, 1)
+	s.Equal(task.GetDomainID(), taskInfo[0].GetDomainID())
+}
+
+func (s *taskAckManagerSuite) TestReadTasksWithBatchSize_Error() {
+	s.mockExecutionMgr.On("GetReplicationTasks", mock.Anything).Return(nil, errors.New("test"))
+
+	taskInfo, hasMore, err := s.ackManager.readTasksWithBatchSize(0, 1)
+	s.Error(err)
+	s.False(hasMore)
+	s.Len(taskInfo, 0)
+}
+
+func (s *taskAckManagerSuite) TestIsNewRunNDCEnabled_True() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(&persistence.VersionHistories{})
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	isNDC, err := s.ackManager.isNewRunNDCEnabled(
+		context.Background(),
+		domainID,
+		workflowID,
+		runID,
+	)
+	s.NoError(err)
+	s.True(isNDC)
+}
+
+func (s *taskAckManagerSuite) TestIsNewRunNDCEnabled_False() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(nil)
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	isNDC, err := s.ackManager.isNewRunNDCEnabled(
+		context.Background(),
+		domainID,
+		workflowID,
+		runID,
+	)
+	s.NoError(err)
+	s.False(isNDC)
+}
+
+func (s *taskAckManagerSuite) TestGetVersionHistoryItems_Error() {
+	_, _, err := getVersionHistoryItems(nil, 0, 0)
+	s.Error(err)
+}
+
+func (s *taskAckManagerSuite) TestGetVersionHistoryItems_OK() {
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 1,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	versionHistory, branchToken, err := getVersionHistoryItems(versionHistories, 1, 1)
+	s.NoError(err)
+	s.Equal(versionHistories.Histories[0].GetBranchToken(), branchToken)
+	s.Equal(versionHistories.Histories[0].Items[0].GetVersion(), versionHistory[0].GetVersion())
+}
+
+func (s *taskAckManagerSuite) TestGetEventsBlob_OK() {
+	branchToken := []byte{}
+	firstEventID := int64(1)
+	nextEventID := int64(2)
+
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(
+		&persistence.ReadRawHistoryBranchResponse{
+			HistoryEventBlobs: []*persistence.DataBlob{
+				{
+					Encoding: common.EncodingTypeJSON,
+					Data:     []byte{},
+				},
+			},
+			Size: 1,
+		}, nil)
+	_, err := s.ackManager.getEventsBlob(branchToken, firstEventID, nextEventID)
+	s.NoError(err)
+}
+
+func (s *taskAckManagerSuite) TestGetEventsBlob_Errors() {
+	branchToken := []byte{}
+	firstEventID := int64(1)
+	nextEventID := int64(2)
+
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(
+		&persistence.ReadRawHistoryBranchResponse{
+			HistoryEventBlobs: []*persistence.DataBlob{},
+			Size:              0,
+		}, nil)
+	_, err := s.ackManager.getEventsBlob(branchToken, firstEventID, nextEventID)
+	s.Error(err)
+
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(
+		&persistence.ReadRawHistoryBranchResponse{
+			HistoryEventBlobs: []*persistence.DataBlob{
+				{
+					Encoding: common.EncodingTypeJSON,
+					Data:     []byte{},
+				},
+				{
+					Encoding: common.EncodingTypeJSON,
+					Data:     []byte{},
+				},
+			},
+			Size: 2,
+		}, nil)
+	_, err = s.ackManager.getEventsBlob(branchToken, firstEventID, nextEventID)
+	s.Error(err)
+
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(nil, errors.New("test"))
+	_, err = s.ackManager.getEventsBlob(branchToken, firstEventID, nextEventID)
+	s.Error(err)
+}
+
+func (s *taskAckManagerSuite) TestProcessReplication_OK() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:   domainID,
+		WorkflowID: workflowID,
+		RunID:      runID,
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(nil).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(nil, false).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	_, err := s.ackManager.processReplication(
+		context.Background(),
+		false,
+		taskInfo,
+		func(
+			activityInfo *persistence.ActivityInfo,
+			versionHistories *persistence.VersionHistories,
+		) (*replicator.ReplicationTask, error) {
+			_, release, err := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+				domainID,
+				workflow.WorkflowExecution{
+					WorkflowId: common.StringPtr(workflowID),
+					RunId:      common.StringPtr(runID),
+				},
+			)
+			s.NoError(err)
+			defer release(nil)
+			return nil, nil
+		},
+	)
+	s.NoError(err)
+	_, release, err = s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	s.NoError(err)
+	release(nil)
+
+	_, err = s.ackManager.processReplication(
+		context.Background(),
+		true,
+		taskInfo,
+		func(
+			activityInfo *persistence.ActivityInfo,
+			versionHistories *persistence.VersionHistories,
+		) (*replicator.ReplicationTask, error) {
+			_, release, err := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+				domainID,
+				workflow.WorkflowExecution{
+					WorkflowId: common.StringPtr(workflowID),
+					RunId:      common.StringPtr(runID),
+				},
+			)
+			s.NoError(err)
+			release(nil)
+			return nil, nil
+		},
+	)
+	s.NoError(err)
+	_, release, err = s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	s.NoError(err)
+	release(nil)
+}
+
+func (s *taskAckManagerSuite) TestProcessReplication_Error() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:   domainID,
+		WorkflowID: workflowID,
+		RunID:      runID,
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, errors.New("test")).Times(1)
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(nil).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(nil, false).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	_, err := s.ackManager.processReplication(
+		context.Background(),
+		true,
+		taskInfo,
+		func(
+			activityInfo *persistence.ActivityInfo,
+			versionHistories *persistence.VersionHistories,
+		) (*replicator.ReplicationTask, error) {
+			return nil, nil
+		},
+	)
+	s.Error(err)
+	_, release, err = s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	s.NoError(err)
+	release(nil)
+}
+
+func (s *taskAckManagerSuite) TestGenerateFailoverMarkerTask() {
+	domainID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:     domainID,
+		TaskID:       1,
+		Version:      2,
+		CreationTime: 3,
+	}
+	task := s.ackManager.generateFailoverMarkerTask(taskInfo)
+	s.Equal(task.GetSourceTaskId(), int64(1))
+	s.NotNil(task.GetFailoverMarkerAttributes())
+	s.Equal(replicator.ReplicationTaskTypeFailoverMarker, task.GetTaskType())
+	s.Equal(domainID, task.GetFailoverMarkerAttributes().GetDomainID())
+	s.Equal(int64(2), task.GetFailoverMarkerAttributes().GetFailoverVersion())
+	s.Equal(int64(3), task.GetFailoverMarkerAttributes().GetCreationTime())
+}
+
+func (s *taskAckManagerSuite) TestGenerateSyncActivityTask_OK() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:   domainID,
+		WorkflowID: workflowID,
+		RunID:      runID,
+	}
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 6,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	activityInfo := &persistence.ActivityInfo{
+		Version:                  1,
+		ScheduleID:               5,
+		ScheduledTime:            time.Now(),
+		StartedID:                6,
+		StartedTime:              time.Now(),
+		DomainID:                 domainID,
+		RequestID:                uuid.New(),
+		LastHeartBeatUpdatedTime: time.Now(),
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(activityInfo, true).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	task, err := s.ackManager.generateSyncActivityTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.NotNil(task)
+	s.NotNil(task.SyncActivityTaskAttributes)
+	s.Equal(replicator.ReplicationTaskTypeSyncActivity, task.GetTaskType())
+	s.Equal(activityInfo.DomainID, task.GetSyncActivityTaskAttributes().GetDomainId())
+	s.Equal(activityInfo.ScheduleID, task.GetSyncActivityTaskAttributes().GetScheduledId())
+	s.Equal(activityInfo.ScheduledTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetScheduledTime())
+	s.Equal(activityInfo.StartedID, task.GetSyncActivityTaskAttributes().GetStartedId())
+	s.Equal(activityInfo.StartedTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetStartedTime())
+	s.Equal(activityInfo.LastHeartBeatUpdatedTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetLastHeartbeatTime())
+	s.Equal(activityInfo.Version, task.GetSyncActivityTaskAttributes().GetVersion())
+	s.Equal(versionHistories.Histories[0].ToThrift(), task.GetSyncActivityTaskAttributes().GetVersionHistory())
+}
+
+func (s *taskAckManagerSuite) TestGenerateSyncActivityTask_Empty() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:   domainID,
+		WorkflowID: workflowID,
+		RunID:      runID,
+	}
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 6,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(nil, false).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	task, err := s.ackManager.generateSyncActivityTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.Nil(task)
+}
+
+func (s *taskAckManagerSuite) TestGenerateHistoryReplicationTask() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		DomainID:     domainID,
+		WorkflowID:   workflowID,
+		RunID:        runID,
+		FirstEventID: 6,
+		Version:      1,
+	}
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 6,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(nil, false).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(
+		&persistence.ReadRawHistoryBranchResponse{
+			HistoryEventBlobs: []*persistence.DataBlob{
+				{
+					Encoding: common.EncodingTypeJSON,
+					Data:     []byte{},
+				},
+			},
+			Size: 1,
+		},
+		nil,
+	)
+
+	task, err := s.ackManager.generateHistoryReplicationTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.NotNil(task)
+	s.NotNil(task.HistoryTaskV2Attributes)
+	s.Equal(replicator.ReplicationTaskTypeHistoryV2, task.GetTaskType())
+}
+
+func (s *taskAckManagerSuite) TestToReplicationTask_FailoverMarker() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		TaskType:     persistence.ReplicationTaskTypeFailoverMarker,
+		DomainID:     domainID,
+		WorkflowID:   workflowID,
+		RunID:        runID,
+		FirstEventID: 6,
+		Version:      1,
+	}
+
+	task, err := s.ackManager.toReplicationTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.NotNil(task)
+	s.Equal(replicator.ReplicationTaskTypeFailoverMarker, task.GetTaskType())
+}
+
+func (s *taskAckManagerSuite) TestToReplicationTask_SyncActivity() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		TaskType:   persistence.ReplicationTaskTypeSyncActivity,
+		DomainID:   domainID,
+		WorkflowID: workflowID,
+		RunID:      runID,
+	}
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 6,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	activityInfo := &persistence.ActivityInfo{
+		Version:                  1,
+		ScheduleID:               5,
+		ScheduledTime:            time.Now(),
+		StartedID:                6,
+		StartedTime:              time.Now(),
+		DomainID:                 domainID,
+		RequestID:                uuid.New(),
+		LastHeartBeatUpdatedTime: time.Now(),
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(activityInfo, true).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+
+	task, err := s.ackManager.toReplicationTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.NotNil(task)
+	s.NotNil(task.SyncActivityTaskAttributes)
+	s.Equal(replicator.ReplicationTaskTypeSyncActivity, task.GetTaskType())
+	s.Equal(activityInfo.DomainID, task.GetSyncActivityTaskAttributes().GetDomainId())
+	s.Equal(activityInfo.ScheduleID, task.GetSyncActivityTaskAttributes().GetScheduledId())
+	s.Equal(activityInfo.ScheduledTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetScheduledTime())
+	s.Equal(activityInfo.StartedID, task.GetSyncActivityTaskAttributes().GetStartedId())
+	s.Equal(activityInfo.StartedTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetStartedTime())
+	s.Equal(activityInfo.LastHeartBeatUpdatedTime.UnixNano(), task.GetSyncActivityTaskAttributes().GetLastHeartbeatTime())
+	s.Equal(activityInfo.Version, task.GetSyncActivityTaskAttributes().GetVersion())
+	s.Equal(versionHistories.Histories[0].ToThrift(), task.GetSyncActivityTaskAttributes().GetVersionHistory())
+}
+
+func (s *taskAckManagerSuite) TestToReplicationTask_History() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	taskInfo := &persistence.ReplicationTaskInfo{
+		TaskType:     persistence.ReplicationTaskTypeHistory,
+		DomainID:     domainID,
+		WorkflowID:   workflowID,
+		RunID:        runID,
+		FirstEventID: 6,
+		Version:      1,
+	}
+	versionHistories := &persistence.VersionHistories{
+		CurrentVersionHistoryIndex: 0,
+		Histories: []*persistence.VersionHistory{
+			{
+				BranchToken: []byte{1},
+				Items: []*persistence.VersionHistoryItem{
+					{
+						EventID: 6,
+						Version: 1,
+					},
+				},
+			},
+		},
+	}
+	workflowContext, release, _ := s.ackManager.executionCache.GetOrCreateWorkflowExecutionForBackground(
+		domainID,
+		workflow.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+	)
+	workflowContext.SetWorkflowExecution(s.mockMutableState)
+	release(nil)
+
+	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).AnyTimes()
+	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
+	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
+	s.mockMutableState.EXPECT().GetActivityInfo(gomock.Any()).Return(nil, false).AnyTimes()
+	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
+		&persistence.DomainInfo{ID: domainID, Name: "domainName"},
+		&persistence.DomainConfig{Retention: 1},
+		&persistence.DomainReplicationConfig{
+			ActiveClusterName: cluster.TestCurrentClusterName,
+			Clusters: []*persistence.ClusterReplicationConfig{
+				{ClusterName: cluster.TestCurrentClusterName},
+				{ClusterName: cluster.TestAlternativeClusterName},
+			},
+		},
+		1,
+		nil,
+	), nil).AnyTimes()
+	s.mockHistoryMgr.On("ReadRawHistoryBranch", mock.Anything).Return(
+		&persistence.ReadRawHistoryBranchResponse{
+			HistoryEventBlobs: []*persistence.DataBlob{
+				{
+					Encoding: common.EncodingTypeJSON,
+					Data:     []byte{},
+				},
+			},
+			Size: 1,
+		},
+		nil,
+	)
+
+	task, err := s.ackManager.generateHistoryReplicationTask(context.Background(), taskInfo)
+	s.NoError(err)
+	s.NotNil(task)
+	s.NotNil(task.HistoryTaskV2Attributes)
+	s.Equal(replicator.ReplicationTaskTypeHistoryV2, task.GetTaskType())
+}
+
+func (s *taskAckManagerSuite) TestGetTasks() {
+	domainID := uuid.New()
+	workflowID := uuid.New()
+	runID := uuid.New()
+	clusterName := "cluster"
+	taskInfo := &persistence.ReplicationTaskInfo{
+		TaskType:     persistence.ReplicationTaskTypeFailoverMarker,
+		DomainID:     domainID,
+		WorkflowID:   workflowID,
+		RunID:        runID,
+		FirstEventID: 6,
+		Version:      1,
+	}
+	s.mockExecutionMgr.On("GetReplicationTasks", mock.Anything).Return(&persistence.GetReplicationTasksResponse{
+		Tasks:         []*persistence.ReplicationTaskInfo{taskInfo},
+		NextPageToken: []byte{1},
+	}, nil)
+	s.mockShard.Resource.ShardMgr.On("UpdateShard", mock.Anything).Return(nil)
+
+	_, err := s.ackManager.GetTasks(context.Background(), clusterName, 10)
+	s.NoError(err)
+	ackLevel := s.mockShard.GetClusterReplicationLevel(clusterName)
+	s.Equal(int64(10), ackLevel)
+}
diff --git a/service/history/replicatorQueueProcessor_test.go b/service/history/replicatorQueueProcessor_test.go
deleted file mode 100644
index 9266a1975..000000000
--- a/service/history/replicatorQueueProcessor_test.go
+++ /dev/null
@@ -1,507 +0,0 @@
-// Copyright (c) 2017 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-package history
-
-import (
-	"testing"
-	"time"
-
-	"github.com/golang/mock/gomock"
-	"github.com/pborman/uuid"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
-
-	"github.com/uber/cadence/.gen/go/replicator"
-	"github.com/uber/cadence/.gen/go/shared"
-	workflow "github.com/uber/cadence/.gen/go/shared"
-	"github.com/uber/cadence/common"
-	"github.com/uber/cadence/common/cache"
-	"github.com/uber/cadence/common/cluster"
-	"github.com/uber/cadence/common/log"
-	"github.com/uber/cadence/common/mocks"
-	"github.com/uber/cadence/common/persistence"
-	"github.com/uber/cadence/service/history/config"
-	"github.com/uber/cadence/service/history/constants"
-	"github.com/uber/cadence/service/history/execution"
-	"github.com/uber/cadence/service/history/shard"
-)
-
-type (
-	replicatorQueueProcessorSuite struct {
-		suite.Suite
-		*require.Assertions
-
-		controller          *gomock.Controller
-		mockShard           *shard.TestContext
-		mockDomainCache     *cache.MockDomainCache
-		mockMutableState    *execution.MockMutableState
-		mockClusterMetadata *cluster.MockMetadata
-
-		mockExecutionMgr *mocks.ExecutionManager
-		mockHistoryV2Mgr *mocks.HistoryV2Manager
-		mockProducer     *mocks.KafkaProducer
-
-		logger log.Logger
-
-		replicatorQueueProcessor *replicatorQueueProcessorImpl
-	}
-)
-
-func TestReplicatorQueueProcessorSuite(t *testing.T) {
-	s := new(replicatorQueueProcessorSuite)
-	suite.Run(t, s)
-}
-
-func (s *replicatorQueueProcessorSuite) SetupSuite() {
-
-}
-
-func (s *replicatorQueueProcessorSuite) TearDownSuite() {
-
-}
-
-func (s *replicatorQueueProcessorSuite) SetupTest() {
-	s.Assertions = require.New(s.T())
-
-	s.controller = gomock.NewController(s.T())
-	s.mockMutableState = execution.NewMockMutableState(s.controller)
-
-	s.mockShard = shard.NewTestContext(
-		s.controller,
-		&persistence.ShardInfo{
-			ShardID:          0,
-			RangeID:          1,
-			TransferAckLevel: 0,
-		},
-		config.NewForTest(),
-	)
-
-	s.mockProducer = &mocks.KafkaProducer{}
-	s.mockDomainCache = s.mockShard.Resource.DomainCache
-	s.mockExecutionMgr = s.mockShard.Resource.ExecutionMgr
-	s.mockHistoryV2Mgr = s.mockShard.Resource.HistoryMgr
-	s.mockClusterMetadata = s.mockShard.Resource.ClusterMetadata
-	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
-	s.mockClusterMetadata.EXPECT().IsGlobalDomainEnabled().Return(true).AnyTimes()
-
-	s.logger = s.mockShard.GetLogger()
-	executionCache := execution.NewCache(s.mockShard)
-
-	s.replicatorQueueProcessor = newReplicatorQueueProcessor(
-		s.mockShard, executionCache, s.mockProducer, s.mockExecutionMgr, s.mockHistoryV2Mgr, s.logger,
-	).(*replicatorQueueProcessorImpl)
-}
-
-func (s *replicatorQueueProcessorSuite) TearDownTest() {
-	s.controller.Finish()
-	s.mockShard.Finish(s.T())
-	s.mockProducer.AssertExpectations(s.T())
-}
-
-func (s *replicatorQueueProcessorSuite) TestSyncActivity_WorkflowMissing() {
-	domainName := "some random domain name"
-	domainID := constants.TestDomainID
-	workflowID := "some random workflow ID"
-	runID := uuid.New()
-	scheduleID := int64(144)
-	taskID := int64(1444)
-	task := &persistence.ReplicationTaskInfo{
-		TaskType:    persistence.ReplicationTaskTypeSyncActivity,
-		TaskID:      taskID,
-		DomainID:    domainID,
-		WorkflowID:  workflowID,
-		RunID:       runID,
-		ScheduledID: scheduleID,
-	}
-	s.mockExecutionMgr.On("CompleteReplicationTask", &persistence.CompleteReplicationTaskRequest{TaskID: taskID}).Return(nil).Once()
-	s.mockExecutionMgr.On("GetWorkflowExecution", &persistence.GetWorkflowExecutionRequest{
-		DomainID: domainID,
-		Execution: shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	}).Return(nil, &shared.EntityNotExistsError{})
-	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: domainID, Name: domainName},
-		&persistence.DomainConfig{Retention: 1},
-		&persistence.DomainReplicationConfig{
-			ActiveClusterName: cluster.TestCurrentClusterName,
-			Clusters: []*persistence.ClusterReplicationConfig{
-				{ClusterName: cluster.TestCurrentClusterName},
-				{ClusterName: cluster.TestAlternativeClusterName},
-			},
-		},
-		1234,
-		nil,
-	), nil).AnyTimes()
-
-	_, err := s.replicatorQueueProcessor.process(newTaskInfo(nil, task, s.logger, s.mockShard.GetTimeSource().Now()))
-	s.Nil(err)
-}
-
-func (s *replicatorQueueProcessorSuite) TestSyncActivity_WorkflowCompleted() {
-	domainName := "some random domain name"
-	domainID := constants.TestDomainID
-	workflowID := "some random workflow ID"
-	runID := uuid.New()
-	scheduleID := int64(144)
-	taskID := int64(1444)
-	version := int64(2333)
-	task := &persistence.ReplicationTaskInfo{
-		TaskType:    persistence.ReplicationTaskTypeSyncActivity,
-		TaskID:      taskID,
-		DomainID:    domainID,
-		WorkflowID:  workflowID,
-		RunID:       runID,
-		ScheduledID: scheduleID,
-	}
-	s.mockExecutionMgr.On("CompleteReplicationTask", &persistence.CompleteReplicationTaskRequest{TaskID: taskID}).Return(nil).Once()
-
-	context, release, _ := s.replicatorQueueProcessor.executionCache.GetOrCreateWorkflowExecutionForBackground(
-		domainID,
-		shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	)
-
-	context.SetWorkflowExecution(s.mockMutableState)
-	release(nil)
-	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
-	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes()
-	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: domainID, Name: domainName},
-		&persistence.DomainConfig{Retention: 1},
-		&persistence.DomainReplicationConfig{
-			ActiveClusterName: cluster.TestCurrentClusterName,
-			Clusters: []*persistence.ClusterReplicationConfig{
-				{ClusterName: cluster.TestCurrentClusterName},
-				{ClusterName: cluster.TestAlternativeClusterName},
-			},
-		},
-		version,
-		nil,
-	), nil).AnyTimes()
-
-	_, err := s.replicatorQueueProcessor.process(newTaskInfo(nil, task, s.logger, s.mockShard.GetTimeSource().Now()))
-	s.Nil(err)
-}
-
-func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityCompleted() {
-	domainName := "some random domain name"
-	domainID := constants.TestDomainID
-	workflowID := "some random workflow ID"
-	runID := uuid.New()
-	scheduleID := int64(144)
-	taskID := int64(1444)
-	version := int64(2333)
-	task := &persistence.ReplicationTaskInfo{
-		TaskType:    persistence.ReplicationTaskTypeSyncActivity,
-		TaskID:      taskID,
-		DomainID:    domainID,
-		WorkflowID:  workflowID,
-		RunID:       runID,
-		ScheduledID: scheduleID,
-	}
-	s.mockExecutionMgr.On("CompleteReplicationTask", &persistence.CompleteReplicationTaskRequest{TaskID: taskID}).Return(nil).Once()
-
-	context, release, _ := s.replicatorQueueProcessor.executionCache.GetOrCreateWorkflowExecutionForBackground(
-		domainID,
-		shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	)
-
-	context.SetWorkflowExecution(s.mockMutableState)
-	release(nil)
-	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
-	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
-	s.mockMutableState.EXPECT().GetActivityInfo(scheduleID).Return(nil, false).AnyTimes()
-	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: domainID, Name: domainName},
-		&persistence.DomainConfig{Retention: 1},
-		&persistence.DomainReplicationConfig{
-			ActiveClusterName: cluster.TestCurrentClusterName,
-			Clusters: []*persistence.ClusterReplicationConfig{
-				{ClusterName: cluster.TestCurrentClusterName},
-				{ClusterName: cluster.TestAlternativeClusterName},
-			},
-		},
-		version,
-		nil,
-	), nil).AnyTimes()
-
-	_, err := s.replicatorQueueProcessor.process(newTaskInfo(nil, task, s.logger, s.mockShard.GetTimeSource().Now()))
-	s.Nil(err)
-}
-
-func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRetry() {
-	domainName := "some random domain name"
-	domainID := constants.TestDomainID
-	workflowID := "some random workflow ID"
-	runID := uuid.New()
-	scheduleID := int64(144)
-	taskID := int64(1444)
-	version := int64(2333)
-	task := &persistence.ReplicationTaskInfo{
-		TaskType:    persistence.ReplicationTaskTypeSyncActivity,
-		TaskID:      taskID,
-		DomainID:    domainID,
-		WorkflowID:  workflowID,
-		RunID:       runID,
-		ScheduledID: scheduleID,
-	}
-	s.mockExecutionMgr.On("CompleteReplicationTask", &persistence.CompleteReplicationTaskRequest{TaskID: taskID}).Return(nil).Once()
-
-	context, release, _ := s.replicatorQueueProcessor.executionCache.GetOrCreateWorkflowExecutionForBackground(
-		domainID,
-		shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	)
-
-	context.SetWorkflowExecution(s.mockMutableState)
-	release(nil)
-
-	activityVersion := int64(333)
-	activityScheduleID := scheduleID
-	activityScheduledTime := time.Now()
-	activityStartedID := common.EmptyEventID
-	activityStartedTime := time.Time{}
-	activityHeartbeatTime := time.Time{}
-	activityAttempt := int32(16384)
-	activityDetails := []byte("some random activity progress")
-	activityLastFailureReason := "some random reason"
-	activityLastWorkerIdentity := "some random worker identity"
-	activityLastFailureDetails := []byte("some random failure details")
-	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
-	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
-	s.mockMutableState.EXPECT().GetActivityInfo(scheduleID).Return(&persistence.ActivityInfo{
-		Version:                  activityVersion,
-		ScheduleID:               activityScheduleID,
-		ScheduledTime:            activityScheduledTime,
-		StartedID:                activityStartedID,
-		StartedTime:              activityStartedTime,
-		LastHeartBeatUpdatedTime: activityHeartbeatTime,
-		Details:                  activityDetails,
-		Attempt:                  activityAttempt,
-		LastFailureReason:        activityLastFailureReason,
-		LastWorkerIdentity:       activityLastWorkerIdentity,
-		LastFailureDetails:       activityLastFailureDetails,
-	}, true).AnyTimes()
-	versionHistory := &persistence.VersionHistory{
-		BranchToken: []byte{},
-		Items: []*persistence.VersionHistoryItem{
-			{
-				EventID: scheduleID,
-				Version: 333,
-			},
-		},
-	}
-	versionHistories := &persistence.VersionHistories{
-		CurrentVersionHistoryIndex: 0,
-		Histories: []*persistence.VersionHistory{
-			versionHistory,
-		},
-	}
-	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
-	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: domainID, Name: domainName},
-		&persistence.DomainConfig{Retention: 1},
-		&persistence.DomainReplicationConfig{
-			ActiveClusterName: cluster.TestCurrentClusterName,
-			Clusters: []*persistence.ClusterReplicationConfig{
-				{ClusterName: cluster.TestCurrentClusterName},
-				{ClusterName: cluster.TestAlternativeClusterName},
-			},
-		},
-		version,
-		nil,
-	), nil).AnyTimes()
-
-	s.mockProducer.On("Publish", &replicator.ReplicationTask{
-		TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeSyncActivity),
-		SyncActivityTaskAttributes: &replicator.SyncActivityTaskAttributes{
-			DomainId:           common.StringPtr(domainID),
-			WorkflowId:         common.StringPtr(workflowID),
-			RunId:              common.StringPtr(runID),
-			Version:            common.Int64Ptr(activityVersion),
-			ScheduledId:        common.Int64Ptr(activityScheduleID),
-			ScheduledTime:      common.Int64Ptr(activityScheduledTime.UnixNano()),
-			StartedId:          common.Int64Ptr(activityStartedID),
-			StartedTime:        nil,
-			LastHeartbeatTime:  common.Int64Ptr(activityHeartbeatTime.UnixNano()),
-			Details:            activityDetails,
-			Attempt:            common.Int32Ptr(activityAttempt),
-			LastFailureReason:  common.StringPtr(activityLastFailureReason),
-			LastWorkerIdentity: common.StringPtr(activityLastWorkerIdentity),
-			LastFailureDetails: activityLastFailureDetails,
-			VersionHistory:     versionHistory.ToThrift(),
-		},
-	}).Return(nil).Once()
-
-	_, err := s.replicatorQueueProcessor.process(newTaskInfo(nil, task, s.logger, s.mockShard.GetTimeSource().Now()))
-	s.Nil(err)
-}
-
-func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRunning() {
-	domainName := "some random domain name"
-	domainID := constants.TestDomainID
-	workflowID := "some random workflow ID"
-	runID := uuid.New()
-	scheduleID := int64(144)
-	taskID := int64(1444)
-	version := int64(2333)
-	task := &persistence.ReplicationTaskInfo{
-		TaskType:    persistence.ReplicationTaskTypeSyncActivity,
-		TaskID:      taskID,
-		DomainID:    domainID,
-		WorkflowID:  workflowID,
-		RunID:       runID,
-		ScheduledID: scheduleID,
-	}
-	s.mockExecutionMgr.On("CompleteReplicationTask", &persistence.CompleteReplicationTaskRequest{TaskID: taskID}).Return(nil).Once()
-
-	context, release, _ := s.replicatorQueueProcessor.executionCache.GetOrCreateWorkflowExecutionForBackground(
-		domainID,
-		shared.WorkflowExecution{
-			WorkflowId: common.StringPtr(workflowID),
-			RunId:      common.StringPtr(runID),
-		},
-	)
-
-	context.SetWorkflowExecution(s.mockMutableState)
-	release(nil)
-
-	activityVersion := int64(333)
-	activityScheduleID := scheduleID
-	activityScheduledTime := time.Now()
-	activityStartedID := activityScheduleID + 1
-	activityStartedTime := activityScheduledTime.Add(time.Minute)
-	activityHeartbeatTime := activityStartedTime.Add(time.Minute)
-	activityAttempt := int32(16384)
-	activityDetails := []byte("some random activity progress")
-	activityLastFailureReason := "some random reason"
-	activityLastWorkerIdentity := "some random worker identity"
-	activityLastFailureDetails := []byte("some random failure details")
-	s.mockMutableState.EXPECT().StartTransaction(gomock.Any()).Return(false, nil).Times(1)
-	s.mockMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
-	s.mockMutableState.EXPECT().GetActivityInfo(scheduleID).Return(&persistence.ActivityInfo{
-		Version:                  activityVersion,
-		ScheduleID:               activityScheduleID,
-		ScheduledTime:            activityScheduledTime,
-		StartedID:                activityStartedID,
-		StartedTime:              activityStartedTime,
-		LastHeartBeatUpdatedTime: activityHeartbeatTime,
-		Details:                  activityDetails,
-		Attempt:                  activityAttempt,
-		LastFailureReason:        activityLastFailureReason,
-		LastWorkerIdentity:       activityLastWorkerIdentity,
-		LastFailureDetails:       activityLastFailureDetails,
-	}, true).AnyTimes()
-	versionHistory := &persistence.VersionHistory{
-		BranchToken: []byte{},
-		Items: []*persistence.VersionHistoryItem{
-			{
-				EventID: scheduleID,
-				Version: 333,
-			},
-		},
-	}
-	versionHistories := &persistence.VersionHistories{
-		CurrentVersionHistoryIndex: 0,
-		Histories: []*persistence.VersionHistory{
-			versionHistory,
-		},
-	}
-	s.mockMutableState.EXPECT().GetVersionHistories().Return(versionHistories).AnyTimes()
-	s.mockDomainCache.EXPECT().GetDomainByID(domainID).Return(cache.NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: domainID, Name: domainName},
-		&persistence.DomainConfig{Retention: 1},
-		&persistence.DomainReplicationConfig{
-			ActiveClusterName: cluster.TestCurrentClusterName,
-			Clusters: []*persistence.ClusterReplicationConfig{
-				{ClusterName: cluster.TestCurrentClusterName},
-				{ClusterName: cluster.TestAlternativeClusterName},
-			},
-		},
-		version,
-		nil,
-	), nil).AnyTimes()
-	s.mockProducer.On("Publish", &replicator.ReplicationTask{
-		TaskType: replicator.ReplicationTaskType.Ptr(replicator.ReplicationTaskTypeSyncActivity),
-		SyncActivityTaskAttributes: &replicator.SyncActivityTaskAttributes{
-			DomainId:           common.StringPtr(domainID),
-			WorkflowId:         common.StringPtr(workflowID),
-			RunId:              common.StringPtr(runID),
-			Version:            common.Int64Ptr(activityVersion),
-			ScheduledId:        common.Int64Ptr(activityScheduleID),
-			ScheduledTime:      common.Int64Ptr(activityScheduledTime.UnixNano()),
-			StartedId:          common.Int64Ptr(activityStartedID),
-			StartedTime:        common.Int64Ptr(activityStartedTime.UnixNano()),
-			LastHeartbeatTime:  common.Int64Ptr(activityHeartbeatTime.UnixNano()),
-			Details:            activityDetails,
-			Attempt:            common.Int32Ptr(activityAttempt),
-			LastFailureReason:  common.StringPtr(activityLastFailureReason),
-			LastWorkerIdentity: common.StringPtr(activityLastWorkerIdentity),
-			LastFailureDetails: activityLastFailureDetails,
-			VersionHistory:     versionHistory.ToThrift(),
-		},
-	}).Return(nil).Once()
-
-	_, err := s.replicatorQueueProcessor.process(newTaskInfo(nil, task, s.logger, s.mockShard.GetTimeSource().Now()))
-	s.Nil(err)
-}
-
-func (s *replicatorQueueProcessorSuite) TestPaginateHistoryWithShardID() {
-	firstEventID := int64(133)
-	nextEventID := int64(134)
-	pageSize := 1
-	shardID := common.IntPtr(1)
-
-	req := &persistence.ReadHistoryBranchRequest{
-		BranchToken:   []byte("asd"),
-		MinEventID:    firstEventID,
-		MaxEventID:    nextEventID,
-		PageSize:      pageSize,
-		NextPageToken: []byte{},
-		ShardID:       shardID,
-	}
-	s.mockHistoryV2Mgr.On("ReadHistoryBranch", req).Return(&persistence.ReadHistoryBranchResponse{
-		HistoryEvents: []*workflow.HistoryEvent{
-			{
-				EventId: common.Int64Ptr(int64(1)),
-			},
-		},
-		NextPageToken:    []byte{},
-		Size:             1,
-		LastFirstEventID: nextEventID,
-	}, nil).Once()
-	hEvents, bEvents, token, size, err := persistence.PaginateHistory(s.mockHistoryV2Mgr, false, []byte("asd"),
-		firstEventID, nextEventID, []byte{}, pageSize, shardID)
-	s.NotNil(hEvents)
-	s.NotNil(bEvents)
-	s.NotNil(token)
-	s.Equal(1, size)
-	s.NoError(err)
-}