Skip to content

Commit e60d6bd

Browse files
yycpttyycptt
authored
Persistence Context Part 3: Namespace Replication Queue (#2638)
1 parent a0c9001 commit e60d6bd

14 files changed

+227
-157
lines changed

‎common/namespace/dlqMessageHandler.go‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -73,12 +73,13 @@ func (d *dlqMessageHandlerImpl) Read(
7373
pageToken []byte,
7474
) ([]*replicationspb.ReplicationTask, []byte, error) {
7575

76-
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel()
76+
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel(ctx)
7777
if err != nil {
7878
return nil, nil, err
7979
}
8080

8181
return d.namespaceReplicationQueue.GetMessagesFromDLQ(
82+
ctx,
8283
ackLevel,
8384
lastMessageID,
8485
pageSize,
@@ -92,19 +93,21 @@ func (d *dlqMessageHandlerImpl) Purge(
9293
lastMessageID int64,
9394
) error {
9495

95-
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel()
96+
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel(ctx)
9697
if err != nil {
9798
return err
9899
}
99100

100101
if err := d.namespaceReplicationQueue.RangeDeleteMessagesFromDLQ(
102+
ctx,
101103
ackLevel,
102104
lastMessageID,
103105
); err != nil {
104106
return err
105107
}
106108

107109
if err := d.namespaceReplicationQueue.UpdateDLQAckLevel(
110+
ctx,
108111
lastMessageID,
109112
); err != nil {
110113
d.logger.Error("Failed to update DLQ ack level after purging messages", tag.Error(err))
@@ -121,12 +124,13 @@ func (d *dlqMessageHandlerImpl) Merge(
121124
pageToken []byte,
122125
) ([]byte, error) {
123126

124-
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel()
127+
ackLevel, err := d.namespaceReplicationQueue.GetDLQAckLevel(ctx)
125128
if err != nil {
126129
return nil, err
127130
}
128131

129132
messages, token, err := d.namespaceReplicationQueue.GetMessagesFromDLQ(
133+
ctx,
130134
ackLevel,
131135
lastMessageID,
132136
pageSize,
@@ -153,13 +157,14 @@ func (d *dlqMessageHandlerImpl) Merge(
153157
}
154158

155159
if err := d.namespaceReplicationQueue.RangeDeleteMessagesFromDLQ(
160+
ctx,
156161
ackLevel,
157162
ackedMessageID,
158163
); err != nil {
159164
d.logger.Error("failed to delete merged tasks on merging namespace DLQ message", tag.Error(err))
160165
return nil, err
161166
}
162-
if err := d.namespaceReplicationQueue.UpdateDLQAckLevel(ackedMessageID); err != nil {
167+
if err := d.namespaceReplicationQueue.UpdateDLQAckLevel(ctx, ackedMessageID); err != nil {
163168
d.logger.Error("failed to update ack level on merging namespace DLQ message", tag.Error(err))
164169
}
165170

‎common/namespace/dlqMessageHandler_test.go‎

Lines changed: 36 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -95,8 +95,8 @@ func (s *dlqMessageHandlerSuite) TestReadMessages() {
9595
SourceTaskId: 1,
9696
},
9797
}
98-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
99-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
98+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
99+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
100100
Return(tasks, nil, nil)
101101

102102
resp, token, err := s.dlqMessageHandler.Read(context.Background(), lastMessageID, pageSize, pageToken)
@@ -118,8 +118,8 @@ func (s *dlqMessageHandlerSuite) TestReadMessages_ThrowErrorOnGetDLQAckLevel() {
118118
},
119119
}
120120
testError := fmt.Errorf("test")
121-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(int64(-1), testError)
122-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
121+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(int64(-1), testError)
122+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
123123
Return(tasks, nil, nil).Times(0)
124124

125125
_, _, err := s.dlqMessageHandler.Read(context.Background(), lastMessageID, pageSize, pageToken)
@@ -134,8 +134,8 @@ func (s *dlqMessageHandlerSuite) TestReadMessages_ThrowErrorOnReadMessages() {
134134
pageToken := []byte{}
135135

136136
testError := fmt.Errorf("test")
137-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
138-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
137+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
138+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
139139
Return(nil, nil, testError)
140140

141141
_, _, err := s.dlqMessageHandler.Read(context.Background(), lastMessageID, pageSize, pageToken)
@@ -147,9 +147,9 @@ func (s *dlqMessageHandlerSuite) TestPurgeMessages() {
147147
ackLevel := int64(10)
148148
lastMessageID := int64(20)
149149

150-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
151-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(ackLevel, lastMessageID).Return(nil)
152-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(lastMessageID).Return(nil)
150+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
151+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID).Return(nil)
152+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), lastMessageID).Return(nil)
153153
err := s.dlqMessageHandler.Purge(context.Background(), lastMessageID)
154154

155155
s.NoError(err)
@@ -159,9 +159,9 @@ func (s *dlqMessageHandlerSuite) TestPurgeMessages_ThrowErrorOnGetDLQAckLevel()
159159
lastMessageID := int64(20)
160160
testError := fmt.Errorf("test")
161161

162-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(int64(-1), testError)
163-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), gomock.Any()).Return(nil).Times(0)
164-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any()).Times(0)
162+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(int64(-1), testError)
163+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(0)
164+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), gomock.Any()).Times(0)
165165
err := s.dlqMessageHandler.Purge(context.Background(), lastMessageID)
166166

167167
s.Equal(testError, err)
@@ -172,9 +172,9 @@ func (s *dlqMessageHandlerSuite) TestPurgeMessages_ThrowErrorOnPurgeMessages() {
172172
lastMessageID := int64(20)
173173
testError := fmt.Errorf("test")
174174

175-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
176-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(ackLevel, lastMessageID).Return(testError)
177-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any()).Times(0)
175+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
176+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID).Return(testError)
177+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), gomock.Any()).Times(0)
178178
err := s.dlqMessageHandler.Purge(context.Background(), lastMessageID)
179179

180180
s.Equal(testError, err)
@@ -200,12 +200,12 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages() {
200200
},
201201
},
202202
}
203-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
204-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
203+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
204+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
205205
Return(tasks, nil, nil)
206206
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute).Return(nil)
207-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(messageID).Return(nil)
208-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(ackLevel, messageID).Return(nil)
207+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), messageID).Return(nil)
208+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), ackLevel, messageID).Return(nil)
209209

210210
token, err := s.dlqMessageHandler.Merge(context.Background(), lastMessageID, pageSize, pageToken)
211211
s.NoError(err)
@@ -231,12 +231,12 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages_ThrowErrorOnGetDLQAckLevel()
231231
},
232232
},
233233
}
234-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(int64(-1), testError)
235-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
234+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(int64(-1), testError)
235+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
236236
Return(tasks, nil, nil).Times(0)
237237
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(0)
238-
s.mockReplicationQueue.EXPECT().DeleteMessageFromDLQ(gomock.Any()).Times(0)
239-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any()).Times(0)
238+
s.mockReplicationQueue.EXPECT().DeleteMessageFromDLQ(gomock.Any(), gomock.Any()).Times(0)
239+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), gomock.Any()).Times(0)
240240

241241
token, err := s.dlqMessageHandler.Merge(context.Background(), lastMessageID, pageSize, pageToken)
242242
s.Equal(testError, err)
@@ -250,12 +250,12 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages_ThrowErrorOnGetDLQMessages()
250250
pageToken := []byte{}
251251
testError := fmt.Errorf("test")
252252

253-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
254-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
253+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
254+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
255255
Return(nil, nil, testError)
256256
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(0)
257-
s.mockReplicationQueue.EXPECT().DeleteMessageFromDLQ(gomock.Any()).Times(0)
258-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any()).Times(0)
257+
s.mockReplicationQueue.EXPECT().DeleteMessageFromDLQ(gomock.Any(), gomock.Any()).Times(0)
258+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), gomock.Any()).Times(0)
259259

260260
token, err := s.dlqMessageHandler.Merge(context.Background(), lastMessageID, pageSize, pageToken)
261261
s.Equal(testError, err)
@@ -292,8 +292,8 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages_ThrowErrorOnHandleReceivingTa
292292
},
293293
},
294294
}
295-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
296-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
295+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
296+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
297297
Return(tasks, nil, nil)
298298
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute1).Return(nil)
299299
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute2).Return(testError)
@@ -333,12 +333,12 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages_ThrowErrorOnDeleteMessages()
333333
},
334334
},
335335
}
336-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
337-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
336+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
337+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
338338
Return(tasks, nil, nil)
339339
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute1).Return(nil)
340340
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute2).Return(nil)
341-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(ackLevel, messageID2).Return(testError)
341+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), ackLevel, messageID2).Return(testError)
342342

343343
token, err := s.dlqMessageHandler.Merge(context.Background(), lastMessageID, pageSize, pageToken)
344344
s.Error(err)
@@ -365,12 +365,12 @@ func (s *dlqMessageHandlerSuite) TestMergeMessages_IgnoreErrorOnUpdateDLQAckLeve
365365
},
366366
},
367367
}
368-
s.mockReplicationQueue.EXPECT().GetDLQAckLevel().Return(ackLevel, nil)
369-
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(ackLevel, lastMessageID, pageSize, pageToken).
368+
s.mockReplicationQueue.EXPECT().GetDLQAckLevel(gomock.Any()).Return(ackLevel, nil)
369+
s.mockReplicationQueue.EXPECT().GetMessagesFromDLQ(gomock.Any(), ackLevel, lastMessageID, pageSize, pageToken).
370370
Return(tasks, nil, nil)
371371
s.mockReplicationTaskExecutor.EXPECT().Execute(gomock.Any(), namespaceAttribute).Return(nil)
372-
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(ackLevel, messageID).Return(nil)
373-
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(messageID).Return(testError)
372+
s.mockReplicationQueue.EXPECT().RangeDeleteMessagesFromDLQ(gomock.Any(), ackLevel, messageID).Return(nil)
373+
s.mockReplicationQueue.EXPECT().UpdateDLQAckLevel(gomock.Any(), messageID).Return(testError)
374374

375375
token, err := s.dlqMessageHandler.Merge(context.Background(), lastMessageID, pageSize, pageToken)
376376
s.NoError(err)

‎common/namespace/handler.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -268,6 +268,7 @@ func (d *HandlerImpl) RegisterNamespace(
268268
}
269269

270270
err = d.namespaceReplicator.HandleTransmissionTask(
271+
ctx,
271272
enumsspb.NAMESPACE_OPERATION_CREATE,
272273
namespaceRequest.Namespace.Info,
273274
namespaceRequest.Namespace.Config,
@@ -570,8 +571,16 @@ func (d *HandlerImpl) UpdateNamespace(
570571
}
571572
}
572573

573-
err = d.namespaceReplicator.HandleTransmissionTask(enumsspb.NAMESPACE_OPERATION_UPDATE,
574-
info, config, replicationConfig, configVersion, failoverVersion, isGlobalNamespace)
574+
err = d.namespaceReplicator.HandleTransmissionTask(
575+
ctx,
576+
enumsspb.NAMESPACE_OPERATION_UPDATE,
577+
info,
578+
config,
579+
replicationConfig,
580+
configVersion,
581+
failoverVersion,
582+
isGlobalNamespace,
583+
)
575584
if err != nil {
576585
return nil, err
577586
}

‎common/namespace/handler_GlobalNamespaceEnabled_MasterCluster_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -441,7 +441,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledMasterClusterSuite) TestRegisterG
441441
})
442442
}
443443

444-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil).Times(0)
444+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil).Times(0)
445445

446446
retention := 1 * time.Hour * 24
447447
registerResp, err := s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
@@ -503,7 +503,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledMasterClusterSuite) TestRegisterG
503503
data := map[string]string{"some random key": "some random value"}
504504
isGlobalNamespace := true
505505

506-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil)
506+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil)
507507

508508
registerResp, err := s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
509509
Namespace: namespace,
@@ -569,7 +569,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledMasterClusterSuite) TestUpdateGet
569569
s.True(len(clusters) > 1)
570570
isGlobalNamespace := true
571571

572-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil).Times(2)
572+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil).Times(2)
573573

574574
registerResp, err := s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
575575
Namespace: namespace,
@@ -653,7 +653,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledMasterClusterSuite) TestUpdateGet
653653
s.True(len(clusters) > 1)
654654
isGlobalNamespace := true
655655

656-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil).Times(2)
656+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil).Times(2)
657657

658658
registerResp, err := s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
659659
Namespace: namespace,
@@ -761,7 +761,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledMasterClusterSuite) TestUpdateGet
761761
s.True(len(clusters) > 1)
762762
isGlobalNamespace := true
763763

764-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil).Times(2)
764+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil).Times(2)
765765

766766
registerResp, err := s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
767767
Namespace: namespace,

‎common/namespace/handler_GlobalNamespaceEnabled_NotMasterCluster_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -519,7 +519,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledNotMasterClusterSuite) TestUpdate
519519
})
520520
s.NoError(err)
521521

522-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil)
522+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil)
523523
resp, err := s.handler.UpdateNamespace(context.Background(), &workflowservice.UpdateNamespaceRequest{
524524
Namespace: namespace,
525525
})
@@ -578,7 +578,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledNotMasterClusterSuite) TestUpdate
578578
})
579579
s.NoError(err)
580580

581-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil)
581+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil)
582582
updateResp, err := s.handler.UpdateNamespace(context.Background(), &workflowservice.UpdateNamespaceRequest{
583583
Namespace: namespace,
584584
UpdateInfo: &namespacepb.UpdateNamespaceInfo{
@@ -686,7 +686,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledNotMasterClusterSuite) TestUpdate
686686
s.Equal(isGlobalNamespace, isGlobalNamespace)
687687
}
688688

689-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil)
689+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil)
690690

691691
updateResp, err := s.handler.UpdateNamespace(context.Background(), &workflowservice.UpdateNamespaceRequest{
692692
Namespace: namespace,

‎common/namespace/handler_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -295,7 +295,7 @@ func (s *namespaceHandlerCommonSuite) TestListNamespace() {
295295
ClusterName: clusterName,
296296
})
297297
}
298-
s.mockProducer.EXPECT().Publish(gomock.Any()).Return(nil)
298+
s.mockProducer.EXPECT().Publish(gomock.Any(), gomock.Any()).Return(nil)
299299
registerResp, err = s.handler.RegisterNamespace(context.Background(), &workflowservice.RegisterNamespaceRequest{
300300
Namespace: namespace2,
301301
Description: description2,

‎common/namespace/transmissionTaskHandler.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
package namespace
2626

2727
import (
28+
"context"
29+
2830
namespacepb "go.temporal.io/api/namespace/v1"
2931
replicationpb "go.temporal.io/api/replication/v1"
3032

@@ -41,6 +43,7 @@ type (
4143
// Replicator is the interface which can replicate the namespace
4244
Replicator interface {
4345
HandleTransmissionTask(
46+
ctx context.Context,
4447
namespaceOperation enumsspb.NamespaceOperation,
4548
info *persistencespb.NamespaceInfo,
4649
config *persistencespb.NamespaceConfig,
@@ -70,6 +73,7 @@ func NewNamespaceReplicator(
7073

7174
// HandleTransmissionTask handle transmission of the namespace replication task
7275
func (namespaceReplicator *namespaceReplicatorImpl) HandleTransmissionTask(
76+
ctx context.Context,
7377
namespaceOperation enumsspb.NamespaceOperation,
7478
info *persistencespb.NamespaceInfo,
7579
config *persistencespb.NamespaceConfig,
@@ -116,6 +120,7 @@ func (namespaceReplicator *namespaceReplicatorImpl) HandleTransmissionTask(
116120
}
117121

118122
return namespaceReplicator.namespaceReplicationQueue.Publish(
123+
ctx,
119124
&replicationspb.ReplicationTask{
120125
TaskType: taskType,
121126
Attributes: task,

0 commit comments

Comments
 (0)