Add a sentinel interface for terminal task errors (#5365)

## What changed?
I added a sentinel interface for errors that we will use to immediately
send a task to the DLQ

## Why?
Having a sentinel (or marker) interface for this cleans up our
implementation and let's us trivially extend it to support new kinds of
terminal errors as we discover them

## How did you test it?

Existing tests

## Potential risks
None

## Is hotfix candidate?
No

common/persistence/serialization/serializer.go

@@ -295,6 +295,10 @@ func (e *UnknownEncodingTypeError) Error() string {
 	)
 }
 
+// IsTerminalTaskError informs our task processing subsystem that it is impossible
+// to retry this error
+func (e *UnknownEncodingTypeError) IsTerminalTaskError() {}
+
 // NewSerializationError returns a SerializationError
 func NewSerializationError(
 	encodingType enumspb.EncodingType,
@@ -333,6 +337,10 @@ func (e *DeserializationError) Unwrap() error {
 	return e.wrappedErr
 }
 
+// IsTerminalTaskError informs our task processing subsystem that it is impossible to
+// retry this error and that the task should be sent to a DLQ
+func (e *DeserializationError) IsTerminalTaskError() {}
+
 func (t *serializerImpl) ShardInfoToBlob(info *persistencespb.ShardInfo, encodingType enumspb.EncodingType) (*commonpb.DataBlob, error) {
 	return ProtoEncodeBlob(info, encodingType)
 }

common/persistence/visibility/store/elasticsearch/client/client.go

@@ -75,6 +75,7 @@ type (
 		IndexPutSettings(ctx context.Context, indexName string, bodyString string) (bool, error)
 		IndexGetSettings(ctx context.Context, indexName string) (map[string]*elastic.IndicesGetSettingsResponse, error)
 		PutMapping(ctx context.Context, index string, mapping map[string]enumspb.IndexedValueType) (bool, error)
+		Ping(ctx context.Context) error
 	}
 
 	// SearchParameters holds all required and optional parameters for executing a search.

common/persistence/visibility/store/elasticsearch/client/client_mock.go

@@ -626,6 +626,20 @@ func (mr *MockIntegrationTestsClientMockRecorder) IndexPutTemplate(ctx, template
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexPutTemplate", reflect.TypeOf((*MockIntegrationTestsClient)(nil).IndexPutTemplate), ctx, templateName, bodyString)
 }
 
+// Ping mocks base method.
+func (m *MockIntegrationTestsClient) Ping(ctx context.Context) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Ping", ctx)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Ping indicates an expected call of Ping.
+func (mr *MockIntegrationTestsClientMockRecorder) Ping(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ping", reflect.TypeOf((*MockIntegrationTestsClient)(nil).Ping), ctx)
+}
+
 // PutMapping mocks base method.
 func (m *MockIntegrationTestsClient) PutMapping(ctx context.Context, index string, mapping map[string]v1.IndexedValueType) (bool, error) {
 	m.ctrl.T.Helper()

common/persistence/visibility/store/elasticsearch/client/client_v7.go

@@ -396,6 +396,12 @@ func (c *clientImpl) Delete(ctx context.Context, indexName string, docID string,
 	return err
 }
 
+// Ping returns whether or not the elastic search cluster is available
+func (c *clientImpl) Ping(ctx context.Context) error {
+	_, _, err := c.esClient.Ping(c.url.String()).Do(ctx)
+	return err
+}
+
 func getLoggerOptions(logLevel string, logger log.Logger) []elastic.ClientOptionFunc {
 	switch {
 	case strings.EqualFold(logLevel, "trace"):

service/history/queues/executable.go

@@ -47,7 +47,6 @@ import (
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
-	"go.temporal.io/server/common/persistence/serialization"
 	ctasks "go.temporal.io/server/common/tasks"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/history/consts"
@@ -84,6 +83,12 @@ type (
 	ExecutorWrapper interface {
 		Wrap(delegate Executor) Executor
 	}
+
+	// TerminalErrors are errors which cannot be retried and should not be scheduled again.
+	// Tasks should be enqueued to a DLQ immediately if an error implements this interface.
+	TerminalTaskError interface {
+		IsTerminalTaskError()
+	}
 )
 
 var (
@@ -384,17 +389,17 @@ func (e *executableImpl) HandleErr(err error) (retErr error) {
 	}
 
 	// TODO: expand on the errors that should be considered terminal
-	if errors.As(err, new(*serialization.DeserializationError)) ||
-		errors.As(err, new(*serialization.UnknownEncodingTypeError)) {
-		// likely due to data corruption, emit logs, metrics & drop the task by return nil so that
-		// task will be marked as completed, or send it to the DLQ if that is enabled.
+	if errors.As(err, new(TerminalTaskError)) {
+		// Terminal errors are likely due to data corruption.
+		// Drop the task by returning nil so that task will be marked as completed,
+		// or send it to the DLQ if that is enabled.
 		metrics.TaskCorruptionCounter.With(e.taggedMetricsHandler).Record(1)
 		if e.dlqEnabled() {
 			e.logger.Error("Marking task as terminally failed, will send to DLQ", tag.Error(err))
 			e.terminalFailureCause = err
 			return fmt.Errorf("%w: %v", ErrTerminalTaskFailure, err)
 		}
-		e.logger.Error("Drop task due to serialization error", tag.Error(err))
+		e.logger.Error("Dropping task due to terminal error", tag.Error(err))
 		return nil
 	}
 

service/history/queues/executable_mock.go

@@ -484,3 +484,38 @@ func (mr *MockExecutorWrapperMockRecorder) Wrap(delegate interface{}) *gomock.Ca
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Wrap", reflect.TypeOf((*MockExecutorWrapper)(nil).Wrap), delegate)
 }
+
+// MockTerminalTaskError is a mock of TerminalTaskError interface.
+type MockTerminalTaskError struct {
+	ctrl     *gomock.Controller
+	recorder *MockTerminalTaskErrorMockRecorder
+}
+
+// MockTerminalTaskErrorMockRecorder is the mock recorder for MockTerminalTaskError.
+type MockTerminalTaskErrorMockRecorder struct {
+	mock *MockTerminalTaskError
+}
+
+// NewMockTerminalTaskError creates a new mock instance.
+func NewMockTerminalTaskError(ctrl *gomock.Controller) *MockTerminalTaskError {
+	mock := &MockTerminalTaskError{ctrl: ctrl}
+	mock.recorder = &MockTerminalTaskErrorMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTerminalTaskError) EXPECT() *MockTerminalTaskErrorMockRecorder {
+	return m.recorder
+}
+
+// IsTerminalTaskError mocks base method.
+func (m *MockTerminalTaskError) IsTerminalTaskError() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "IsTerminalTaskError")
+}
+
+// IsTerminalTaskError indicates an expected call of IsTerminalTaskError.
+func (mr *MockTerminalTaskErrorMockRecorder) IsTerminalTaskError() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsTerminalTaskError", reflect.TypeOf((*MockTerminalTaskError)(nil).IsTerminalTaskError))
+}

tests/test_cluster.go

@@ -33,6 +33,7 @@ import (
 	"os"
 	"path"
 	"testing"
+	"time"
 
 	"github.com/pborman/uuid"
 	"go.uber.org/fx"
@@ -45,6 +46,7 @@ import (
 	"go.temporal.io/server/common/archiver"
 	"go.temporal.io/server/common/archiver/filestore"
 	"go.temporal.io/server/common/archiver/provider"
+	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/config"
 	"go.temporal.io/server/common/dynamicconfig"
@@ -335,9 +337,24 @@ func NewClusterWithPersistenceTestBaseFactory(t *testing.T, options *TestCluster
 }
 
 func setupIndex(esConfig *esclient.Config, logger log.Logger) error {
-	esClient, err := esclient.NewFunctionalTestsClient(esConfig, logger)
+	var esClient esclient.IntegrationTestsClient
+	op := func() error {
+		var err error
+		esClient, err = esclient.NewFunctionalTestsClient(esConfig, logger)
+		if err != nil {
+			return err
+		}
+
+		return esClient.Ping(context.TODO())
+	}
+
+	err := backoff.ThrottleRetry(
+		op,
+		backoff.NewExponentialRetryPolicy(time.Second).WithExpirationInterval(time.Minute),
+		nil,
+	)
 	if err != nil {
-		return err
+		logger.Fatal("Failed to connect to elasticsearch", tag.Error(err))
 	}
 
 	exists, err := esClient.IndexExists(context.Background(), esConfig.GetVisibilityIndex())