mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Alex ShtinandGitHub
8e8f1b0f7c
## What changed?
<!-- Describe what has changed in this PR -->
Remove license header from every file. Because it is really hard to
follow in this PR here is the summary:
1. License header is removed from all `.go` and `.proto` files
:fireworks::fireworks:🎆.
2. `LICENSE` file in the root directory has only Temporal and Uber
copyrights.
3. 5 other `LICENSE` files added to the packages which have copyrights
different from Temporal and Uber: Datadog, Xargin, "Mat Ryer, Tyler
Bunnell and contributors".
4. `license_file` flag is removed from all code generation tools.
5. `copyright_file` flag is removed from `go:generate mockgen`
directive.
6. All copyright related targets are removed from `Makefile`.
7. Updated Temporal copyright year to 2025 everywhere.
## Why?
<!-- Tell your future self why have you made these changes -->
I double checked with legal department that it is not needed to have
license header in every file. One file per repo is enough. I put all
copyrights to the root `LICENSE` file and removed header from all other
files. Also updated tools and `Makefile`.
291 lines
9.6 KiB
Go
291 lines
9.6 KiB
Go
package replication
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/pborman/uuid"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
"go.temporal.io/server/api/adminservice/v1"
|
|
"go.temporal.io/server/api/adminservicemock/v1"
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
historyspb "go.temporal.io/server/api/history/v1"
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
replicationspb "go.temporal.io/server/api/replication/v1"
|
|
"go.temporal.io/server/client"
|
|
"go.temporal.io/server/common/cluster"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/persistence"
|
|
"go.temporal.io/server/common/resourcetest"
|
|
"go.temporal.io/server/service/history/configs"
|
|
deletemanager "go.temporal.io/server/service/history/deletemanager"
|
|
"go.temporal.io/server/service/history/shard"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
"go.temporal.io/server/service/history/tests"
|
|
wcache "go.temporal.io/server/service/history/workflow/cache"
|
|
"go.uber.org/mock/gomock"
|
|
)
|
|
|
|
type (
|
|
dlqHandlerSuite struct {
|
|
suite.Suite
|
|
*require.Assertions
|
|
controller *gomock.Controller
|
|
|
|
mockResource *resourcetest.Test
|
|
mockShard *shard.ContextTest
|
|
config *configs.Config
|
|
mockClientBean *client.MockBean
|
|
adminClient *adminservicemock.MockAdminServiceClient
|
|
clusterMetadata *cluster.MockMetadata
|
|
executionManager *persistence.MockExecutionManager
|
|
shardManager *persistence.MockShardManager
|
|
taskExecutor *MockTaskExecutor
|
|
taskExecutors map[string]TaskExecutor
|
|
sourceCluster string
|
|
|
|
replicationMessageHandler *dlqHandlerImpl
|
|
}
|
|
)
|
|
|
|
func TestDLQHandlerSuite(t *testing.T) {
|
|
s := new(dlqHandlerSuite)
|
|
suite.Run(t, s)
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) SetupSuite() {
|
|
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) TearDownSuite() {
|
|
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
s.controller = gomock.NewController(s.T())
|
|
|
|
s.mockShard = shard.NewTestContext(
|
|
s.controller,
|
|
&persistencespb.ShardInfo{
|
|
ShardId: 0,
|
|
RangeId: 1,
|
|
ReplicationDlqAckLevel: map[string]int64{cluster.TestAlternativeClusterName: persistence.EmptyQueueMessageID},
|
|
},
|
|
tests.NewDynamicConfig(),
|
|
)
|
|
s.mockResource = s.mockShard.Resource
|
|
s.mockClientBean = s.mockResource.ClientBean
|
|
s.adminClient = s.mockResource.RemoteAdminClient
|
|
s.clusterMetadata = s.mockResource.ClusterMetadata
|
|
s.executionManager = s.mockResource.ExecutionMgr
|
|
s.shardManager = s.mockResource.ShardMgr
|
|
s.config = tests.NewDynamicConfig()
|
|
s.clusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
|
|
s.clusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes()
|
|
s.taskExecutors = make(map[string]TaskExecutor)
|
|
s.taskExecutor = NewMockTaskExecutor(s.controller)
|
|
s.sourceCluster = cluster.TestAlternativeClusterName
|
|
s.taskExecutors[s.sourceCluster] = s.taskExecutor
|
|
|
|
s.replicationMessageHandler = newDLQHandler(
|
|
s.mockShard,
|
|
deletemanager.NewMockDeleteManager(s.controller),
|
|
wcache.NewMockCache(s.controller),
|
|
s.mockClientBean,
|
|
s.taskExecutors,
|
|
func(params TaskExecutorParams) TaskExecutor {
|
|
return NewTaskExecutor(
|
|
params.RemoteCluster,
|
|
params.Shard,
|
|
params.HistoryResender,
|
|
params.DeleteManager,
|
|
params.WorkflowCache,
|
|
)
|
|
},
|
|
)
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) TearDownTest() {
|
|
s.controller.Finish()
|
|
s.mockShard.StopForTest()
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) TestReadMessages_OK() {
|
|
ctx := context.Background()
|
|
|
|
namespaceID := uuid.New()
|
|
workflowID := uuid.New()
|
|
runID := uuid.New()
|
|
taskID := int64(12345)
|
|
version := int64(2333)
|
|
firstEventID := int64(144)
|
|
nextEventID := int64(233)
|
|
|
|
lastMessageID := int64(1394)
|
|
pageSize := 1
|
|
pageToken := []byte("some random token")
|
|
dbResp := &persistence.GetHistoryTasksResponse{
|
|
Tasks: []tasks.Task{&tasks.HistoryReplicationTask{
|
|
WorkflowKey: definition.NewWorkflowKey(
|
|
namespaceID,
|
|
workflowID,
|
|
runID,
|
|
),
|
|
Version: version,
|
|
FirstEventID: firstEventID,
|
|
NextEventID: nextEventID,
|
|
TaskID: taskID,
|
|
}},
|
|
NextPageToken: pageToken,
|
|
}
|
|
|
|
remoteTask := &replicationspb.ReplicationTask{
|
|
TaskType: enumsspb.REPLICATION_TASK_TYPE_HISTORY_TASK,
|
|
SourceTaskId: taskID,
|
|
Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
|
|
HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
|
|
NamespaceId: namespaceID,
|
|
WorkflowId: workflowID,
|
|
RunId: runID,
|
|
VersionHistoryItems: []*historyspb.VersionHistoryItem{{
|
|
Version: version,
|
|
EventId: nextEventID - 1,
|
|
}},
|
|
Events: &commonpb.DataBlob{},
|
|
},
|
|
},
|
|
}
|
|
|
|
s.executionManager.EXPECT().GetReplicationTasksFromDLQ(gomock.Any(), &persistence.GetReplicationTasksFromDLQRequest{
|
|
GetHistoryTasksRequest: persistence.GetHistoryTasksRequest{
|
|
ShardID: s.mockShard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(persistence.EmptyQueueMessageID + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
BatchSize: pageSize,
|
|
NextPageToken: pageToken,
|
|
},
|
|
SourceClusterName: s.sourceCluster,
|
|
}).Return(dbResp, nil)
|
|
|
|
s.mockClientBean.EXPECT().GetRemoteAdminClient(s.sourceCluster).Return(s.adminClient, nil).AnyTimes()
|
|
s.adminClient.EXPECT().GetDLQReplicationMessages(ctx, gomock.Any()).
|
|
Return(&adminservice.GetDLQReplicationMessagesResponse{
|
|
ReplicationTasks: []*replicationspb.ReplicationTask{remoteTask},
|
|
}, nil)
|
|
taskList, tasksInfo, token, err := s.replicationMessageHandler.GetMessages(ctx, s.sourceCluster, lastMessageID, pageSize, pageToken)
|
|
s.NoError(err)
|
|
s.Equal(pageToken, token)
|
|
s.Equal([]*replicationspb.ReplicationTask{remoteTask}, taskList)
|
|
s.Equal(namespaceID, tasksInfo[0].GetNamespaceId())
|
|
s.Equal(workflowID, tasksInfo[0].GetWorkflowId())
|
|
s.Equal(taskID, tasksInfo[0].GetTaskId())
|
|
s.Equal(version, tasksInfo[0].GetVersion())
|
|
s.Equal(firstEventID, tasksInfo[0].GetFirstEventId())
|
|
s.Equal(nextEventID, tasksInfo[0].GetNextEventId())
|
|
}
|
|
|
|
func (s *dlqHandlerSuite) TestPurgeMessages() {
|
|
lastMessageID := int64(1)
|
|
|
|
s.executionManager.EXPECT().RangeDeleteReplicationTaskFromDLQ(
|
|
gomock.Any(),
|
|
&persistence.RangeDeleteReplicationTaskFromDLQRequest{
|
|
RangeCompleteHistoryTasksRequest: persistence.RangeCompleteHistoryTasksRequest{
|
|
ShardID: s.mockShard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(persistence.EmptyQueueMessageID + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
},
|
|
SourceClusterName: s.sourceCluster,
|
|
}).Return(nil)
|
|
|
|
s.shardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil)
|
|
err := s.replicationMessageHandler.PurgeMessages(context.Background(), s.sourceCluster, lastMessageID)
|
|
s.NoError(err)
|
|
}
|
|
func (s *dlqHandlerSuite) TestMergeMessages() {
|
|
ctx := context.Background()
|
|
|
|
namespaceID := uuid.New()
|
|
workflowID := uuid.New()
|
|
runID := uuid.New()
|
|
taskID := int64(12345)
|
|
version := int64(2333)
|
|
firstEventID := int64(144)
|
|
nextEventID := int64(233)
|
|
|
|
lastMessageID := int64(1394)
|
|
pageSize := 1
|
|
pageToken := []byte("some random token")
|
|
|
|
dbResp := &persistence.GetHistoryTasksResponse{
|
|
Tasks: []tasks.Task{&tasks.HistoryReplicationTask{
|
|
WorkflowKey: definition.NewWorkflowKey(
|
|
namespaceID,
|
|
workflowID,
|
|
runID,
|
|
),
|
|
Version: version,
|
|
FirstEventID: firstEventID,
|
|
NextEventID: nextEventID,
|
|
TaskID: taskID,
|
|
}},
|
|
NextPageToken: pageToken,
|
|
}
|
|
|
|
remoteTask := &replicationspb.ReplicationTask{
|
|
TaskType: enumsspb.REPLICATION_TASK_TYPE_HISTORY_TASK,
|
|
SourceTaskId: taskID,
|
|
Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
|
|
HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
|
|
NamespaceId: namespaceID,
|
|
WorkflowId: workflowID,
|
|
RunId: runID,
|
|
VersionHistoryItems: []*historyspb.VersionHistoryItem{{
|
|
Version: version,
|
|
EventId: nextEventID - 1,
|
|
}},
|
|
Events: &commonpb.DataBlob{},
|
|
},
|
|
},
|
|
}
|
|
|
|
s.executionManager.EXPECT().GetReplicationTasksFromDLQ(gomock.Any(), &persistence.GetReplicationTasksFromDLQRequest{
|
|
GetHistoryTasksRequest: persistence.GetHistoryTasksRequest{
|
|
ShardID: s.mockShard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(persistence.EmptyQueueMessageID + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
BatchSize: pageSize,
|
|
NextPageToken: pageToken,
|
|
},
|
|
SourceClusterName: s.sourceCluster,
|
|
}).Return(dbResp, nil)
|
|
|
|
s.mockClientBean.EXPECT().GetRemoteAdminClient(s.sourceCluster).Return(s.adminClient, nil).AnyTimes()
|
|
s.adminClient.EXPECT().GetDLQReplicationMessages(ctx, gomock.Any()).
|
|
Return(&adminservice.GetDLQReplicationMessagesResponse{
|
|
ReplicationTasks: []*replicationspb.ReplicationTask{remoteTask},
|
|
}, nil)
|
|
s.taskExecutor.EXPECT().Execute(gomock.Any(), remoteTask, true).Return(nil)
|
|
s.executionManager.EXPECT().RangeDeleteReplicationTaskFromDLQ(gomock.Any(), &persistence.RangeDeleteReplicationTaskFromDLQRequest{
|
|
RangeCompleteHistoryTasksRequest: persistence.RangeCompleteHistoryTasksRequest{
|
|
ShardID: s.mockShard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(persistence.EmptyQueueMessageID + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
},
|
|
SourceClusterName: s.sourceCluster,
|
|
}).Return(nil)
|
|
|
|
s.shardManager.EXPECT().UpdateShard(gomock.Any(), gomock.Any()).Return(nil)
|
|
|
|
token, err := s.replicationMessageHandler.MergeMessages(ctx, s.sourceCluster, lastMessageID, pageSize, pageToken)
|
|
s.NoError(err)
|
|
s.Equal(pageToken, token)
|
|
}
|