Skip to content

Add a feature flag for serialization encoding #7148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Aug 21, 2025
Merged

Add a feature flag for serialization encoding #7148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6cc561a
Add thrift_snappy endoding
tubignat Jul 23, 2025
93561a8
Move test payloads into a common file
tubignat Jul 24, 2025
796acb7
Merge branch 'cadence-workflow:master' into master
tubignat Aug 7, 2025
aeba827
Add feature flag for serialization encoding
tubignat Aug 7, 2025
2180910
Fix tests
tubignat Aug 8, 2025
7c4d793
Fix tests
tubignat Aug 11, 2025
8f9097e
Merge branch 'cadence-workflow:master' into master
tubignat Aug 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions common/dynamicconfig/dynamicproperties/constants.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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{
Expand Down
34 changes: 8 additions & 26 deletions common/persistence/client/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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")
}
Expand All @@ -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
}
Expand Down
2 changes: 2 additions & 0 deletions common/persistence/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ type (
EnableHistoryTaskDualWriteMode dynamicproperties.BoolPropertyFn
ReadNoSQLHistoryTaskFromDataBlob dynamicproperties.BoolPropertyFn
ReadNoSQLShardFromDataBlob dynamicproperties.BoolPropertyFn
SerializationEncoding dynamicproperties.StringPropertyFn
}
)

Expand All @@ -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),
}
}
2 changes: 2 additions & 0 deletions common/persistence/persistence-tests/persistenceTestBase.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading