Add a feature flag for serialization encoding (#7148)

* Add thrift_snappy endoding

* Move test payloads into a common file

* Add feature flag for serialization encoding

* Fix tests

* Fix tests

Author: tubignat

common/dynamicconfig/dynamicproperties/constants.go

@@ -2418,6 +2418,12 @@ const (
 
 	// LastStringKey must be the last one in this const group
 	LastStringKey
+
+	// SerializationEncoding is the encoding type for blobs
+	// KeyName: history.serializationEncoding
+	// Value type: String
+	// Default value: "thriftrw"
+	SerializationEncoding
 )
 
 const (
@@ -4911,6 +4917,11 @@ var StringKeys = map[StringKey]DynamicString{
 		Description:  "MatchingShardDistributionMode defines which shard distribution mode should be used",
 		DefaultValue: "hash_ring",
 	},
+	SerializationEncoding: {
+		KeyName:      "history.serializationEncoding",
+		Description:  "SerializationEncoding is the encoding type for blobs",
+		DefaultValue: string(constants.EncodingTypeThriftRW),
+	},
 }
 
 var DurationKeys = map[DurationKey]DynamicDuration{

common/persistence/client/factory.go

@@ -526,12 +526,12 @@ func (f *factoryImpl) init(clusterName string, limiters map[string]quotas.Limite
 	defaultDataStore := Datastore{ratelimit: limiters[f.config.DefaultStore]}
 	switch {
 	case defaultCfg.NoSQL != nil:
-		parser := getParser(f.logger, constants.EncodingTypeThriftRW, constants.EncodingTypeThriftRW)
+		parser := f.getParser()
 		taskSerializer := serialization.NewTaskSerializer(parser)
 		shardedNoSQLConfig := defaultCfg.NoSQL.ConvertToShardedNoSQLConfig()
 		defaultDataStore.factory = nosql.NewFactory(*shardedNoSQLConfig, clusterName, f.logger, f.metricsClient, taskSerializer, parser, f.dc)
 	case defaultCfg.ShardedNoSQL != nil:
-		parser := getParser(f.logger, constants.EncodingTypeThriftRW, constants.EncodingTypeThriftRW)
+		parser := f.getParser()
 		taskSerializer := serialization.NewTaskSerializer(parser)
 		defaultDataStore.factory = nosql.NewFactory(*defaultCfg.ShardedNoSQL, clusterName, f.logger, f.metricsClient, taskSerializer, parser, f.dc)
 	case defaultCfg.SQL != nil:
@@ -543,16 +543,7 @@ func (f *factoryImpl) init(clusterName string, limiters map[string]quotas.Limite
 				string(constants.EncodingTypeThriftRW),
 			}
 		}
-		var decodingTypes []constants.EncodingType
-		for _, dt := range defaultCfg.SQL.DecodingTypes {
-			decodingTypes = append(decodingTypes, constants.EncodingType(dt))
-		}
-		defaultDataStore.factory = sql.NewFactory(
-			*defaultCfg.SQL,
-			clusterName,
-			f.logger,
-			getParser(f.logger, constants.EncodingType(defaultCfg.SQL.EncodingType), decodingTypes...),
-			f.dc)
+		defaultDataStore.factory = sql.NewFactory(*defaultCfg.SQL, clusterName, f.logger, f.getParser(), f.dc)
 	default:
 		f.logger.Fatal("invalid config: one of nosql or sql params must be specified for defaultDataStore")
 	}
@@ -576,32 +567,23 @@ func (f *factoryImpl) init(clusterName string, limiters map[string]quotas.Limite
 	visibilityDataStore := Datastore{ratelimit: limiters[f.config.VisibilityStore]}
 	switch {
 	case visibilityCfg.NoSQL != nil:
-		parser := getParser(f.logger, constants.EncodingTypeThriftRW, constants.EncodingTypeThriftRW)
+		parser := f.getParser()
 		taskSerializer := serialization.NewTaskSerializer(parser)
 		shardedNoSQLConfig := visibilityCfg.NoSQL.ConvertToShardedNoSQLConfig()
 		visibilityDataStore.factory = nosql.NewFactory(*shardedNoSQLConfig, clusterName, f.logger, f.metricsClient, taskSerializer, parser, f.dc)
 	case visibilityCfg.SQL != nil:
-		var decodingTypes []constants.EncodingType
-		for _, dt := range visibilityCfg.SQL.DecodingTypes {
-			decodingTypes = append(decodingTypes, constants.EncodingType(dt))
-		}
-		visibilityDataStore.factory = sql.NewFactory(
-			*visibilityCfg.SQL,
-			clusterName,
-			f.logger,
-			getParser(f.logger, constants.EncodingType(visibilityCfg.SQL.EncodingType), decodingTypes...),
-			f.dc)
+		visibilityDataStore.factory = sql.NewFactory(*visibilityCfg.SQL, clusterName, f.logger, f.getParser(), f.dc)
 	default:
 		f.logger.Fatal("invalid config: one of nosql or sql params must be specified for visibilityStore")
 	}
 
 	f.datastores[storeTypeVisibility] = visibilityDataStore
 }
 
-func getParser(logger log.Logger, encodingType constants.EncodingType, decodingTypes ...constants.EncodingType) serialization.Parser {
-	parser, err := serialization.NewParser(encodingType, decodingTypes...)
+func (f *factoryImpl) getParser() serialization.Parser {
+	parser, err := serialization.NewParser(f.dc)
 	if err != nil {
-		logger.Fatal("failed to construct parser", tag.Error(err))
+		f.logger.Fatal("failed to construct parser", tag.Error(err))
 	}
 	return parser
 }

common/persistence/config.go

@@ -35,6 +35,7 @@ type (
 		EnableHistoryTaskDualWriteMode           dynamicproperties.BoolPropertyFn
 		ReadNoSQLHistoryTaskFromDataBlob         dynamicproperties.BoolPropertyFn
 		ReadNoSQLShardFromDataBlob               dynamicproperties.BoolPropertyFn
+		SerializationEncoding                    dynamicproperties.StringPropertyFn
 	}
 )
 
@@ -48,5 +49,6 @@ func NewDynamicConfiguration(dc *dynamicconfig.Collection) *DynamicConfiguration
 		EnableHistoryTaskDualWriteMode:           dc.GetBoolProperty(dynamicproperties.EnableNoSQLHistoryTaskDualWriteMode),
 		ReadNoSQLHistoryTaskFromDataBlob:         dc.GetBoolProperty(dynamicproperties.ReadNoSQLHistoryTaskFromDataBlob),
 		ReadNoSQLShardFromDataBlob:               dc.GetBoolProperty(dynamicproperties.ReadNoSQLShardFromDataBlob),
+		SerializationEncoding:                    dc.GetStringProperty(dynamicproperties.SerializationEncoding),
 	}
 }

common/persistence/persistence-tests/persistenceTestBase.go

@@ -157,6 +157,7 @@ func NewTestBaseWithNoSQL(t *testing.T, options *TestBaseOptions) *TestBase {
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
 		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
+		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
 	}
 	params := TestBaseParams{
 		DefaultTestCluster:    testCluster,
@@ -188,6 +189,7 @@ func NewTestBaseWithSQL(t *testing.T, options *TestBaseOptions) *TestBase {
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
 		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
+		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
 	}
 	params := TestBaseParams{
 		DefaultTestCluster:    testCluster,