Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow start many replication pollers in one shard #3790

Merged
merged 7 commits into from
Jan 13, 2023
Merged

Allow start many replication pollers in one shard #3790

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
52ab61f
Allow start many replication poller in one shard
yux0 Jan 9, 2023
d5ce7d2
handle test
yux0 Jan 10, 2023
811faa2
Rename method
yux0 Jan 12, 2023
da5bb3a
revert change to use the local shard id for config
yux0 Jan 13, 2023
fe395fd
Merge branch 'master' into start-replication-poller
yux0 Jan 13, 2023
ce06871
fix functional test
yux0 Jan 13, 2023
48d1d0c
rename variables
yux0 Jan 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions service/history/replication/poller_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,12 +31,18 @@ import (
)

type (
pollerManager interface {
getSourceClusterShardIDs(sourceClusterName string) []int32
}

pollerManagerImpl struct {
currentShardId int32
clusterMetadata cluster.Metadata
}
)

var _ pollerManager = (*pollerManagerImpl)(nil)

func newPollerManager(
currentShardId int32,
clusterMetadata cluster.Metadata,
Expand All @@ -47,21 +53,21 @@ func newPollerManager(
}
}

func (p pollerManagerImpl) getPollingShardIDs(remoteClusterName string) []int32 {
func (p pollerManagerImpl) getSourceClusterShardIDs(sourceClusterName string) []int32 {
currentCluster := p.clusterMetadata.GetCurrentClusterName()
allClusters := p.clusterMetadata.GetAllClusterInfo()
currentClusterInfo, ok := allClusters[currentCluster]
if !ok {
panic("Cannot get current cluster info from cluster metadata cache")
}
remoteClusterInfo, ok := allClusters[remoteClusterName]
remoteClusterInfo, ok := allClusters[sourceClusterName]
if !ok {
panic(fmt.Sprintf("Cannot get remote cluster %s info from cluster metadata cache", remoteClusterName))
panic(fmt.Sprintf("Cannot get source cluster %s info from cluster metadata cache", sourceClusterName))
}
return generatePollingShardIDs(p.currentShardId, currentClusterInfo.ShardCount, remoteClusterInfo.ShardCount)
return generateShardIDs(p.currentShardId, currentClusterInfo.ShardCount, remoteClusterInfo.ShardCount)
}

func generatePollingShardIDs(localShardId int32, localShardCount int32, remoteShardCount int32) []int32 {
func generateShardIDs(localShardId int32, localShardCount int32, remoteShardCount int32) []int32 {
var pollingShards []int32
if remoteShardCount <= localShardCount {
if localShardId <= remoteShardCount {
Expand Down
2 changes: 1 addition & 1 deletion service/history/replication/poller_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ func TestGetPollingShardIds(t *testing.T) {
t.Errorf("The code did not panic")
}
}()
shardIDs := generatePollingShardIDs(tt.shardID, tt.localShardCount, tt.remoteShardCount)
shardIDs := generateShardIDs(tt.shardID, tt.localShardCount, tt.remoteShardCount)
assert.Equal(t, tt.expectedShardIDs, shardIDs)
})
}
Expand Down
17 changes: 11 additions & 6 deletions service/history/replication/task_processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,10 @@ type (

// taskProcessorImpl is responsible for processing replication tasks for a shard.
taskProcessorImpl struct {
currentCluster string
status int32

sourceCluster string
status int32
sourceShardID int32
shard shard.Context
historyEngine shard.Engine
historySerializer serialization.Serializer
Expand Down Expand Up @@ -109,6 +110,7 @@ type (

// NewTaskProcessor creates a new replication task processor.
func NewTaskProcessor(
sourceShardID int32,
shard shard.Context,
historyEngine shard.Engine,
config *configs.Config,
Expand All @@ -132,9 +134,9 @@ func NewTaskProcessor(
WithExpirationInterval(config.ReplicationTaskProcessorErrorRetryExpiration(shardID))

return &taskProcessorImpl{
currentCluster: shard.GetClusterMetadata().GetCurrentClusterName(),
sourceCluster: replicationTaskFetcher.getSourceCluster(),
status: common.DaemonStatusInitialized,
sourceShardID: sourceShardID,
sourceCluster: replicationTaskFetcher.getSourceCluster(),
shard: shard,
historyEngine: historyEngine,
historySerializer: eventSerializer,
Expand Down Expand Up @@ -383,6 +385,7 @@ func (p *taskProcessorImpl) convertTaskToDLQTask(
switch replicationTask.TaskType {
case enumsspb.REPLICATION_TASK_TYPE_SYNC_ACTIVITY_TASK:
taskAttributes := replicationTask.GetSyncActivityTaskAttributes()
// TODO: GetShardID will break GetDLQReplicationMessages we need to handle DLQ for cross shard replication.
return &persistence.PutReplicationTaskToDLQRequest{
ShardID: p.shard.GetShardID(),
SourceClusterName: p.sourceCluster,
Expand Down Expand Up @@ -414,6 +417,7 @@ func (p *taskProcessorImpl) convertTaskToDLQTask(
// NOTE: last event vs next event, next event ID is exclusive
nextEventID := lastEvent.GetEventId() + 1

// TODO: GetShardID will break GetDLQReplicationMessages we need to handle DLQ for cross shard replication.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why will it break? Using local shardID for storing DLQ seems the right behavior to me.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. Here is ok. But when hydrate the replication task from the source cluster, it needs to calculate the shard id instead of using this. We will not hit this case in a short term.

return &persistence.PutReplicationTaskToDLQRequest{
ShardID: p.shard.GetShardID(),
SourceClusterName: p.sourceCluster,
Expand Down Expand Up @@ -442,6 +446,7 @@ func (p *taskProcessorImpl) convertTaskToDLQTask(
return nil, err
}

// TODO: GetShardID will break GetDLQReplicationMessages we need to handle DLQ for cross shard replication.
return &persistence.PutReplicationTaskToDLQRequest{
ShardID: p.shard.GetShardID(),
SourceClusterName: p.sourceCluster,
Expand All @@ -464,7 +469,7 @@ func (p *taskProcessorImpl) paginationFn(_ []byte) ([]interface{}, []byte, error
respChan := make(chan *replicationspb.ReplicationMessages, 1)
p.requestChan <- &replicationTaskRequest{
token: &replicationspb.ReplicationToken{
ShardId: p.shard.GetShardID(),
ShardId: p.sourceShardID,
LastProcessedMessageId: p.maxRxProcessedTaskID,
LastProcessedVisibilityTime: &p.maxRxProcessedTimestamp,
LastRetrievedMessageId: p.maxRxReceivedTaskID,
Expand Down Expand Up @@ -499,7 +504,7 @@ func (p *taskProcessorImpl) paginationFn(_ []byte) ([]interface{}, []byte, error
if resp.GetHasMore() {
p.rxTaskBackoff = time.Duration(0)
} else {
p.rxTaskBackoff = p.config.ReplicationTaskProcessorNoTaskRetryWait(p.shard.GetShardID())
p.rxTaskBackoff = p.config.ReplicationTaskProcessorNoTaskRetryWait(p.sourceShardID)
}
return tasks, nil, nil

Expand Down
70 changes: 40 additions & 30 deletions service/history/replication/task_processor_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ package replication

import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
Expand All @@ -49,6 +50,10 @@ import (
wcache "go.temporal.io/server/service/history/workflow/cache"
)

const (
clusterCallbackKey = "%s-%d" // <cluster name>-<polling shard id>
)

type (
// taskProcessorManagerImpl is to manage replication task processors
taskProcessorManagerImpl struct {
Expand All @@ -62,6 +67,7 @@ type (
workflowCache wcache.Cache
resender xdc.NDCHistoryResender
taskExecutorProvider TaskExecutorProvider
taskPollerManager pollerManager
metricsHandler metrics.Handler
logger log.Logger

Expand Down Expand Up @@ -110,6 +116,7 @@ func NewTaskProcessorManager(
metricsHandler: shard.GetMetricsHandler(),
taskProcessors: make(map[string]TaskProcessor),
taskExecutorProvider: taskExecutorProvider,
taskPollerManager: newPollerManager(shard.GetShardID(), shard.GetClusterMetadata()),
minTxAckedTaskID: persistence.EmptyQueueMessageID,
shutdownChan: make(chan struct{}),
}
Expand Down Expand Up @@ -167,36 +174,39 @@ func (r *taskProcessorManagerImpl) handleClusterMetadataUpdate(
if clusterName == currentClusterName {
continue
}
// The metadata triggers a update when the following fields update: 1. Enabled 2. Initial Failover Version 3. Cluster address
// The callback covers three cases:
// Case 1: Remove a cluster Case 2: Add a new cluster Case 3: Refresh cluster metadata.

if processor, ok := r.taskProcessors[clusterName]; ok {
// Case 1 and Case 3
processor.Stop()
delete(r.taskProcessors, clusterName)
}

if clusterInfo := newClusterMetadata[clusterName]; clusterInfo != nil && clusterInfo.Enabled {
// Case 2 and Case 3
fetcher := r.replicationTaskFetcherFactory.GetOrCreateFetcher(clusterName)
replicationTaskProcessor := NewTaskProcessor(
r.shard,
r.engine,
r.config,
r.shard.GetMetricsHandler(),
fetcher,
r.taskExecutorProvider(TaskExecutorParams{
RemoteCluster: clusterName,
Shard: r.shard,
HistoryResender: r.resender,
DeleteManager: r.deleteMgr,
WorkflowCache: r.workflowCache,
}),
r.eventSerializer,
)
replicationTaskProcessor.Start()
r.taskProcessors[clusterName] = replicationTaskProcessor
sourceShardIds := r.taskPollerManager.getSourceClusterShardIDs(clusterName)
for _, sourceShardId := range sourceShardIds {
perShardTaskProcessorKey := fmt.Sprintf(clusterCallbackKey, clusterName, sourceShardId)
// The metadata triggers an update when the following fields update: 1. Enabled 2. Initial Failover Version 3. Cluster address
// The callback covers three cases:
// Case 1: Remove a cluster Case 2: Add a new cluster Case 3: Refresh cluster metadata.
if processor, ok := r.taskProcessors[perShardTaskProcessorKey]; ok {
// Case 1 and Case 3
processor.Stop()
delete(r.taskProcessors, perShardTaskProcessorKey)
}
if clusterInfo := newClusterMetadata[clusterName]; clusterInfo != nil && clusterInfo.Enabled {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what if a cluster exists in both old & new cluster metadata? stop & restart?
what if a cluster only exists in both new cluster metadata? noop?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if a cluster exists in both old and new, it means the metadata is updated. So stop the processor and restart it to load the new metadata. If a cluster only exists in new cluster. It means this cluster is newly added, so we just start a new processor.

// Case 2 and Case 3
fetcher := r.replicationTaskFetcherFactory.GetOrCreateFetcher(clusterName)
replicationTaskProcessor := NewTaskProcessor(
sourceShardId,
r.shard,
r.engine,
r.config,
r.shard.GetMetricsHandler(),
fetcher,
r.taskExecutorProvider(TaskExecutorParams{
RemoteCluster: clusterName,
Shard: r.shard,
HistoryResender: r.resender,
DeleteManager: r.deleteMgr,
WorkflowCache: r.workflowCache,
}),
r.eventSerializer,
)
replicationTaskProcessor.Start()
r.taskProcessors[perShardTaskProcessorKey] = replicationTaskProcessor
}
}
}
}
Expand Down
1 change: 1 addition & 0 deletions service/history/replication/task_processor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ func (s *taskProcessorSuite) SetupTest() {
metricsClient := metrics.NoopMetricsHandler

s.replicationTaskProcessor = NewTaskProcessor(
s.shardID,
s.mockShard,
s.mockEngine,
s.config,
Expand Down
4 changes: 4 additions & 0 deletions tests/test_cluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,10 @@ func NewCluster(options *TestClusterConfig, logger log.Logger) (*TestCluster, er
}
}

clusterInfoMap := make(map[string]cluster.ClusterInformation)
for clusterName, clusterInfo := range clusterMetadataConfig.ClusterInformation {
clusterInfo.ShardCount = options.HistoryConfig.NumHistoryShards
clusterInfoMap[clusterName] = clusterInfo
_, err := testBase.ClusterMetadataManager.SaveClusterMetadata(context.Background(), &persistence.SaveClusterMetadataRequest{
ClusterMetadata: persistencespb.ClusterMetadata{
HistoryShardCount: options.HistoryConfig.NumHistoryShards,
Expand All @@ -185,6 +188,7 @@ func NewCluster(options *TestClusterConfig, logger log.Logger) (*TestCluster, er
return nil, err
}
}
clusterMetadataConfig.ClusterInformation = clusterInfoMap

// This will save custom test search attributes to cluster metadata.
// Actual Elasticsearch fields are created from index template (testdata/es_v7_index_template.json).
Expand Down