Support feature flags in client (#1103)

Author: demirkayaender

.gen/go/shared/shared.go

@@ -47,6 +47,9 @@ type (
 	// Options are optional parameters for Client creation.
 	Options = internal.ClientOptions
 
+	// FeatureFlags define which breaking changes can be enabled for client
+	FeatureFlags = internal.FeatureFlags
+
 	// StartWorkflowOptions configuration parameters for starting a workflow execution.
 	StartWorkflowOptions = internal.StartWorkflowOptions
 

client/client.go

@@ -87,7 +87,7 @@ func TestWorkersTestSuite(t *testing.T) {
 }
 
 // this is the mock for yarpcCallOptions, make sure length are the same
-var callOptions = []interface{}{gomock.Any(), gomock.Any(), gomock.Any()}
+var callOptions = []interface{}{gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()}
 
 func createTestEventWorkflowExecutionStarted(eventID int64, attr *m.WorkflowExecutionStartedEventAttributes) *m.HistoryEvent {
 	return &m.HistoryEvent{

evictiontest/workflow_cache_eviction_test.go

@@ -55,11 +55,13 @@ func (s *activityTestSuite) TearDownTest() {
 }
 
 // this is the mock for yarpcCallOptions, make sure length are the same
-var callOptions = []interface{}{gomock.Any(), gomock.Any(), gomock.Any()}
+var callOptions = []interface{}{gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()}
+
+var featureFlags = FeatureFlags{}
 
 func (s *activityTestSuite) TestActivityHeartbeat() {
 	ctx, cancel := context.WithCancel(context.Background())
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}))
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{serviceInvoker: invoker})
 
 	s.service.EXPECT().RecordActivityTaskHeartbeat(gomock.Any(), gomock.Any(), callOptions...).
@@ -70,7 +72,7 @@ func (s *activityTestSuite) TestActivityHeartbeat() {
 
 func (s *activityTestSuite) TestActivityHeartbeat_InternalError() {
 	ctx, cancel := context.WithCancel(context.Background())
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}))
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker,
 		logger:         getTestLogger(s.T())})
@@ -86,7 +88,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_InternalError() {
 
 func (s *activityTestSuite) TestActivityHeartbeat_CancelRequested() {
 	ctx, cancel := context.WithCancel(context.Background())
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}))
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker,
 		logger:         getTestLogger(s.T())})
@@ -101,7 +103,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_CancelRequested() {
 
 func (s *activityTestSuite) TestActivityHeartbeat_EntityNotExist() {
 	ctx, cancel := context.WithCancel(context.Background())
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}))
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 1, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker,
 		logger:         getTestLogger(s.T())})
@@ -116,7 +118,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_EntityNotExist() {
 
 func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 	ctx, cancel := context.WithCancel(context.Background())
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 2, make(chan struct{}))
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 2, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker,
 		logger:         getTestLogger(s.T())})
@@ -131,7 +133,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 
 	// No HB timeout configured.
 	service2 := workflowservicetest.NewMockClient(s.mockCtrl)
-	invoker2 := newServiceInvoker([]byte("task-token"), "identity", service2, cancel, 0, make(chan struct{}))
+	invoker2 := newServiceInvoker([]byte("task-token"), "identity", service2, cancel, 0, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker2,
 		logger:         getTestLogger(s.T())})
@@ -144,7 +146,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 	// simulate batch picks before expiry.
 	waitCh := make(chan struct{})
 	service3 := workflowservicetest.NewMockClient(s.mockCtrl)
-	invoker3 := newServiceInvoker([]byte("task-token"), "identity", service3, cancel, 2, make(chan struct{}))
+	invoker3 := newServiceInvoker([]byte("task-token"), "identity", service3, cancel, 2, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker3,
 		logger:         getTestLogger(s.T())})
@@ -174,7 +176,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 	// simulate batch picks before expiry, with out any progress specified.
 	waitCh2 := make(chan struct{})
 	service4 := workflowservicetest.NewMockClient(s.mockCtrl)
-	invoker4 := newServiceInvoker([]byte("task-token"), "identity", service4, cancel, 2, make(chan struct{}))
+	invoker4 := newServiceInvoker([]byte("task-token"), "identity", service4, cancel, 2, make(chan struct{}), featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{
 		serviceInvoker: invoker4,
 		logger:         getTestLogger(s.T())})
@@ -198,7 +200,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 func (s *activityTestSuite) TestActivityHeartbeat_WorkerStop() {
 	ctx, cancel := context.WithCancel(context.Background())
 	workerStopChannel := make(chan struct{})
-	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 5, workerStopChannel)
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, cancel, 5, workerStopChannel, featureFlags)
 	ctx = context.WithValue(ctx, activityEnvContextKey, &activityEnvironment{serviceInvoker: invoker})
 
 	heartBeatDetail := "testDetails"

idls

@@ -335,6 +335,7 @@ type (
 		DataConverter      DataConverter
 		Tracer             opentracing.Tracer
 		ContextPropagators []ContextPropagator
+		FeatureFlags       FeatureFlags
 	}
 
 	// StartWorkflowOptions configuration parameters for starting a workflow execution.
@@ -502,6 +503,15 @@ const (
 	WorkflowIDReusePolicyTerminateIfRunning
 )
 
+func getFeatureFlags(options *ClientOptions) FeatureFlags {
+	if options != nil {
+		return FeatureFlags{
+			WorkflowExecutionAlreadyCompletedErrorEnabled: options.FeatureFlags.WorkflowExecutionAlreadyCompletedErrorEnabled,
+		}
+	}
+	return FeatureFlags{}
+}
+
 // NewClient creates an instance of a workflow client
 func NewClient(service workflowserviceclient.Interface, domain string, options *ClientOptions) Client {
 	var identity string
@@ -532,6 +542,7 @@ func NewClient(service workflowserviceclient.Interface, domain string, options *
 	} else {
 		tracer = opentracing.NoopTracer{}
 	}
+
 	return &workflowClient{
 		workflowService:    metrics.NewWorkflowServiceWrapper(service, metricScope),
 		domain:             domain,
@@ -541,6 +552,7 @@ func NewClient(service workflowserviceclient.Interface, domain string, options *
 		dataConverter:      dataConverter,
 		contextPropagators: contextPropagators,
 		tracer:             tracer,
+		featureFlags:       getFeatureFlags(options),
 	}
 }
 
@@ -557,10 +569,12 @@ func NewDomainClient(service workflowserviceclient.Interface, options *ClientOpt
 		metricScope = options.MetricsScope
 	}
 	metricScope = tagScope(metricScope, tagDomain, "domain-client", clientImplHeaderName, clientImplHeaderValue)
+
 	return &domainClient{
 		workflowService: metrics.NewWorkflowServiceWrapper(service, metricScope),
 		metricsScope:    metricScope,
 		identity:        identity,
+		featureFlags:    getFeatureFlags(options),
 	}
 }
 

internal/activity_test.go

@@ -144,6 +144,7 @@ type (
 		workerStopCh       <-chan struct{}
 		contextPropagators []ContextPropagator
 		tracer             opentracing.Tracer
+		featureFlags       FeatureFlags
 	}
 
 	// history wrapper method to help information about events.
@@ -1659,6 +1660,7 @@ func newActivityTaskHandlerWithCustomProvider(
 		workerStopCh:       params.WorkerStopChannel,
 		contextPropagators: params.ContextPropagators,
 		tracer:             params.Tracer,
+		featureFlags:       params.FeatureFlags,
 	}
 }
 
@@ -1674,6 +1676,7 @@ type cadenceInvoker struct {
 	lastDetailsReported   *[]byte     // Details that were reported in the last reporting interval.
 	closeCh               chan struct{}
 	workerStopChannel     <-chan struct{}
+	featureFlags          FeatureFlags
 }
 
 func (i *cadenceInvoker) Heartbeat(details []byte) error {
@@ -1778,7 +1781,7 @@ func (i *cadenceInvoker) internalHeartBeat(details []byte) (bool, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
-	err := recordActivityHeartbeat(ctx, i.service, i.identity, i.taskToken, details)
+	err := recordActivityHeartbeat(ctx, i.service, i.identity, i.taskToken, details, i.featureFlags)
 
 	switch err.(type) {
 	case *CanceledError:
@@ -1813,7 +1816,7 @@ func (i *cadenceInvoker) Close(flushBufferedHeartbeat bool) {
 }
 
 func (i *cadenceInvoker) SignalWorkflow(ctx context.Context, domain, workflowID, runID, signalName string, signalInput []byte) error {
-	return signalWorkflow(ctx, i.service, i.identity, domain, workflowID, runID, signalName, signalInput)
+	return signalWorkflow(ctx, i.service, i.identity, domain, workflowID, runID, signalName, signalInput, i.featureFlags)
 }
 
 func newServiceInvoker(
@@ -1823,6 +1826,7 @@ func newServiceInvoker(
 	cancelHandler func(),
 	heartBeatTimeoutInSec int32,
 	workerStopChannel <-chan struct{},
+	featureFlags FeatureFlags,
 ) ServiceInvoker {
 	return &cadenceInvoker{
 		taskToken:             taskToken,
@@ -1832,6 +1836,7 @@ func newServiceInvoker(
 		heartBeatTimeoutInSec: heartBeatTimeoutInSec,
 		closeCh:               make(chan struct{}),
 		workerStopChannel:     workerStopChannel,
+		featureFlags:          featureFlags,
 	}
 }
 
@@ -1851,7 +1856,7 @@ func (ath *activityTaskHandlerImpl) Execute(taskList string, t *s.PollForActivit
 	canCtx, cancel := context.WithCancel(rootCtx)
 	defer cancel()
 
-	invoker := newServiceInvoker(t.TaskToken, ath.identity, ath.service, cancel, t.GetHeartbeatTimeoutSeconds(), ath.workerStopCh)
+	invoker := newServiceInvoker(t.TaskToken, ath.identity, ath.service, cancel, t.GetHeartbeatTimeoutSeconds(), ath.workerStopCh, ath.featureFlags)
 	defer func() {
 		_, activityCompleted := result.(*s.RespondActivityTaskCompletedRequest)
 		invoker.Close(!activityCompleted) // flush buffered heartbeat if activity was not successfully completed.
@@ -1977,6 +1982,7 @@ func signalWorkflow(
 	runID string,
 	signalName string,
 	signalInput []byte,
+	featureFlags FeatureFlags,
 ) error {
 	request := &s.SignalWorkflowExecutionRequest{
 		Domain: common.StringPtr(domain),
@@ -1991,7 +1997,7 @@ func signalWorkflow(
 
 	return backoff.Retry(ctx,
 		func() error {
-			tchCtx, cancel, opt := newChannelContext(ctx)
+			tchCtx, cancel, opt := newChannelContext(ctx, featureFlags)
 			defer cancel()
 			return service.SignalWorkflowExecution(tchCtx, request, opt...)
 		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
@@ -2002,6 +2008,7 @@ func recordActivityHeartbeat(
 	service workflowserviceclient.Interface,
 	identity string,
 	taskToken, details []byte,
+	featureFlags FeatureFlags,
 ) error {
 	request := &s.RecordActivityTaskHeartbeatRequest{
 		TaskToken: taskToken,
@@ -2011,7 +2018,7 @@ func recordActivityHeartbeat(
 	var heartbeatResponse *s.RecordActivityTaskHeartbeatResponse
 	heartbeatErr := backoff.Retry(ctx,
 		func() error {
-			tchCtx, cancel, opt := newChannelContext(ctx)
+			tchCtx, cancel, opt := newChannelContext(ctx, featureFlags)
 			defer cancel()
 
 			var err error
@@ -2032,6 +2039,7 @@ func recordActivityHeartbeatByID(
 	identity string,
 	domain, workflowID, runID, activityID string,
 	details []byte,
+	featureFlags FeatureFlags,
 ) error {
 	request := &s.RecordActivityTaskHeartbeatByIDRequest{
 		Domain:     common.StringPtr(domain),
@@ -2044,7 +2052,7 @@ func recordActivityHeartbeatByID(
 	var heartbeatResponse *s.RecordActivityTaskHeartbeatResponse
 	heartbeatErr := backoff.Retry(ctx,
 		func() error {
-			tchCtx, cancel, opt := newChannelContext(ctx)
+			tchCtx, cancel, opt := newChannelContext(ctx, featureFlags)
 			defer cancel()
 
 			var err error

internal/client.go

@@ -1265,7 +1265,9 @@ func (t *TaskHandlersTestSuite) TestHeartBeat_NilResponseWithError() {
 		mockService,
 		func() {},
 		0,
-		make(chan struct{}))
+		make(chan struct{}),
+		featureFlags,
+	)
 
 	heartbeatErr := cadenceInvoker.BatchHeartbeat(nil)
 	t.NotNil(heartbeatErr)
@@ -1289,7 +1291,9 @@ func (t *TaskHandlersTestSuite) TestHeartBeat_NilResponseWithDomainNotActiveErro
 		mockService,
 		cancelHandler,
 		0,
-		make(chan struct{}))
+		make(chan struct{}),
+		featureFlags,
+	)
 
 	heartbeatErr := cadenceInvoker.BatchHeartbeat(nil)
 	t.NotNil(heartbeatErr)