[Bug]: Managed transform incompatible schema

[Abacn]

What happened?

Managed transform submitted from older Beam client starting to fail after #34525 due to incompatible parameter introduced.

Issue Priority

Priority: 1

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

cc: @ahmedabu98

[ahmedabu98]

When we generate a configuration schema from schematransforms, we sort the schema lexicographically to make it deterministic. But now that means the new field "skip_config_validation" will be before "transform_identifier".

Changing the index for field "transform_identifier" is causing this issue.

[ahmedabu98]

Actually, we're getting the following error:

INFO 2025-04-30T21:24:09.216460779Z E0430 21:24:09.193300 15 managed_transforms_worker_main.cc:213] Failed to upgrade using the expansion service manager: INTERNAL: Expansion request failed: java.lang.IllegalArgumentException: Config schema provided with the expansion request Fields:
INFO 2025-04-30T21:24:09.216465204Z Field{name=config, description=YAML string config used to build the underlying SchemaTransform., type=STRING, options={{}}}
INFO 2025-04-30T21:24:09.216469481Z Field{name=config_url, description=URL path to the YAML config file used to build the underlying SchemaTransform., type=STRING, options={{}}}
INFO 2025-04-30T21:24:09.216473492Z Field{name=transform_identifier, description=Identifier of the underlying SchemaTransform to discover and instantiate., type=STRING NOT NULL, options={{}}}
INFO 2025-04-30T21:24:09.216477191Z Encoding positions:
INFO 2025-04-30T21:24:09.216480899Z {transform_identifier=2, config_url=1, config=0}
INFO 2025-04-30T21:24:09.216502941Z Options:{{}}UUID: 0443ce49-dd99-478c-861a-8393df4294c2 is not compatible with the config of the Schema transform Fields:
INFO 2025-04-30T21:24:09.216507398Z Field{name=config, description=YAML string config used to build the underlying SchemaTransform., type=STRING, options={{}}}
INFO 2025-04-30T21:24:09.216513612Z Field{name=config_url, description=URL path to the YAML config file used to build the underlying SchemaTransform., type=STRING, options={{}}}
INFO 2025-04-30T21:24:09.216518985Z Field{name=skip_config_validation, description=Skips configuration validation. If unset, the pipeline will fail at construction time if the configuration includes unknown fields or missing required fields., type=BOOLEAN, options={{}}}
INFO 2025-04-30T21:24:09.216522859Z Field{name=transform_identifier, description=Identifier of the underlying SchemaTransform to discover and instantiate., type=STRING NOT NULL, options={{}}}
INFO 2025-04-30T21:24:09.216527841Z Encoding positions:
INFO 2025-04-30T21:24:09.216533340Z {skip_config_validation=2, transform_identifier=3, config_url=1, config=0}

which is coming from here:

beam/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceSchemaTransformProvider.java

Lines 121 to 127 in 9e8e3c3

	if (!configSchemaFromRequest.assignableTo(configSchemaFromProvider)) {
	throw new IllegalArgumentException(
	String.format(
	"Config schema provided with the expansion request %s is not compatible with the "
	+ "config of the Schema transform %s.",
	configSchemaFromRequest, configSchemaFromProvider));
	}

This method, assignableTo(), fails if the two Schemas have different number of fields. #34525 adds a new parameter to this configuration, which creates an imbalance between Beam HEAD and Dataflow service because the service hasn't caught up yet (i.e. with imports)

[ahmedabu98]

When Dataflow service catches up, this new parameter should work fine.. but I think at the same time it would break older SDK versions that don't have the new parameter. Either way, there will be an imbalance according to assignableTo().

I wonder if this logic can be improved to ignore nullable fields? cc @chamikaramj

[Abacn]

Internal compat test back green, close this for now