Add support for Ephemeral TaskLists

internal/activity_task_handler.go

@@ -80,7 +80,7 @@ func newActivityTaskHandlerWithCustomProvider(
 	}
 	return &activityTaskHandlerImpl{
 		clock:              clock,
-		taskListName:       params.TaskList,
+		taskListName:       params.TaskList.GetName(),
 		identity:           params.Identity,
 		service:            service,
 		logger:             params.Logger,

internal/client.go

@@ -656,6 +656,7 @@ func getFeatureFlags(options *ClientOptions) FeatureFlags {
 		return FeatureFlags{
 			WorkflowExecutionAlreadyCompletedErrorEnabled: options.FeatureFlags.WorkflowExecutionAlreadyCompletedErrorEnabled,
 			PollerAutoScalerEnabled:                       options.FeatureFlags.PollerAutoScalerEnabled,
+			EphemeralTaskListsEnabled:                     options.FeatureFlags.EphemeralTaskListsEnabled,
 		}
 	}
 	return FeatureFlags{}

internal/compatibility/enum_test.go

@@ -266,6 +266,7 @@ func TestTaskListKind(t *testing.T) {
 		apiv1.TaskListKind_TASK_LIST_KIND_INVALID,
 		apiv1.TaskListKind_TASK_LIST_KIND_NORMAL,
 		apiv1.TaskListKind_TASK_LIST_KIND_STICKY,
+		apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL,
 	} {
 		assert.Equal(t, item, proto.TaskListKind(thrift.TaskListKind(item)))
 	}

internal/compatibility/proto/enum.go

@@ -35,6 +35,8 @@ func TaskListKind(t *shared.TaskListKind) apiv1.TaskListKind {
 		return apiv1.TaskListKind_TASK_LIST_KIND_NORMAL
 	case shared.TaskListKindSticky:
 		return apiv1.TaskListKind_TASK_LIST_KIND_STICKY
+	case shared.TaskListKindEphemeral:
+		return apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL
 	}
 	panic("unexpected enum value")
 }

internal/compatibility/thrift/enum.go

@@ -34,6 +34,8 @@ func TaskListKind(t apiv1.TaskListKind) *shared.TaskListKind {
 		return shared.TaskListKindNormal.Ptr()
 	case apiv1.TaskListKind_TASK_LIST_KIND_STICKY:
 		return shared.TaskListKindSticky.Ptr()
+	case apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL:
+		return shared.TaskListKindEphemeral.Ptr()
 	}
 	panic("unexpected enum value")
 }

internal/internal_activity.go

@@ -59,6 +59,7 @@ type (
 	activityOptions struct {
 		ActivityID                    *string // Users can choose IDs but our framework makes it optional to decrease the crust.
 		TaskListName                  string
+		TaskListKind                  shared.TaskListKind
 		ScheduleToCloseTimeoutSeconds int32
 		ScheduleToStartTimeoutSeconds int32
 		StartToCloseTimeoutSeconds    int32

internal/internal_event_handlers.go

@@ -121,6 +121,7 @@ type (
 		contextPropagators           []ContextPropagator
 		tracer                       opentracing.Tracer
 		workflowInterceptorFactories []WorkflowInterceptorFactory
+		featureFlags                 FeatureFlags
 	}
 
 	localActivityTask struct {
@@ -205,6 +206,7 @@ func newWorkflowExecutionEventHandler(
 	contextPropagators []ContextPropagator,
 	tracer opentracing.Tracer,
 	workflowInterceptorFactories []WorkflowInterceptorFactory,
+	featureFlags FeatureFlags,
 ) workflowExecutionEventHandler {
 	context := &workflowEnvironmentImpl{
 		workflowInfo:                 workflowInfo,
@@ -222,6 +224,7 @@ func newWorkflowExecutionEventHandler(
 		contextPropagators:           contextPropagators,
 		tracer:                       tracer,
 		workflowInterceptorFactories: workflowInterceptorFactories,
+		featureFlags:                 featureFlags,
 	}
 	context.logger = logger.With(
 		zapcore.Field{Key: tagWorkflowType, Type: zapcore.StringType, String: workflowInfo.WorkflowType.Name},
@@ -472,7 +475,7 @@ func (wc *workflowEnvironmentImpl) ExecuteActivity(parameters executeActivityPar
 	}
 	activityID := scheduleTaskAttr.GetActivityId()
 	scheduleTaskAttr.ActivityType = activityTypePtr(parameters.ActivityType)
-	scheduleTaskAttr.TaskList = common.TaskListPtr(m.TaskList{Name: common.StringPtr(parameters.TaskListName)})
+	scheduleTaskAttr.TaskList = common.TaskListPtr(m.TaskList{Name: common.StringPtr(parameters.TaskListName), Kind: parameters.TaskListKind.Ptr()})
 	scheduleTaskAttr.Input = parameters.Input
 	scheduleTaskAttr.ScheduleToCloseTimeoutSeconds = common.Int32Ptr(parameters.ScheduleToCloseTimeoutSeconds)
 	scheduleTaskAttr.StartToCloseTimeoutSeconds = common.Int32Ptr(parameters.StartToCloseTimeoutSeconds)
@@ -804,6 +807,10 @@ func (wc *workflowEnvironmentImpl) GetWorkflowInterceptors() []WorkflowIntercept
 	return wc.workflowInterceptorFactories
 }
 
+func (wc *workflowEnvironmentImpl) GetFeatureFlags() FeatureFlags {
+	return wc.featureFlags
+}
+
 func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 	event *m.HistoryEvent,
 	isReplay bool,

internal/internal_event_handlers_test.go

@@ -1065,6 +1065,7 @@ func testWorkflowExecutionEventHandler(t *testing.T, registry *registry) *workfl
 		nil,
 		opentracing.NoopTracer{},
 		nil,
+		FeatureFlags{},
 	).(*workflowExecutionEventHandlerImpl)
 }
 

internal/internal_task_handlers.go

@@ -144,6 +144,7 @@ type (
 		tracer                         opentracing.Tracer
 		workflowInterceptorFactories   []WorkflowInterceptorFactory
 		disableStrictNonDeterminism    bool
+		featureFlags                   FeatureFlags
 	}
 
 	activityProvider func(name string) activity
@@ -419,6 +420,7 @@ func newWorkflowTaskHandler(
 		tracer:                         params.Tracer,
 		workflowInterceptorFactories:   params.WorkflowInterceptorChainFactories,
 		disableStrictNonDeterminism:    params.WorkerBugPorts.DisableStrictNonDeterminismCheck,
+		featureFlags:                   params.FeatureFlags,
 	}
 
 	traceLog(func() {
@@ -622,6 +624,7 @@ func (w *workflowExecutionContextImpl) createEventHandler() {
 		w.wth.contextPropagators,
 		w.wth.tracer,
 		w.wth.workflowInterceptorFactories,
+		w.wth.featureFlags,
 	)
 	w.eventHandler.Store(eventHandler)
 }