PDGGK commented on PR #39760:
URL: https://github.com/apache/beam/pull/39760#issuecomment-5373184928
Thanks — I went and looked, since it is a fair thing to want settled before
merging.
**Short answer: the upgrade module is not affected.**
`sdks/java/io/kafka/upgrade/KafkaIOTranslation.java` only registers
translators for the classic transforms — `KafkaIO.Read`, `KafkaIO.Write` and
`KafkaIO.WriteRecords`, under `KAFKA_READ_WITH_METADATA_TRANSFORM_URN_V2` /
`KAFKA_WRITE_TRANSFORM_URN_V2`. It never references
`KafkaWriteSchemaTransformProvider`.
The schema transform's upgrade path is
`KafkaSchemaTransformTranslation.KafkaWriteSchemaTransformTranslator`, and its
whole payload is:
```java
@Override
public Row toConfigRow(KafkaWriteSchemaTransform transform) {
return transform.getConfigurationRow();
}
```
`getConfigurationRow()` serialises `KafkaWriteSchemaTransformConfiguration`,
whose fields are `format`, `topic`, `bootstrapServers`,
`producerConfigUpdates`, `errorHandling`, `fileDescriptorPath`, `messageName`
and `schema` — and that last one is the input message schema for Avro/Proto,
not anything to do with the error output. The schema this PR changes is
attached to the error `PCollection` during `expand()` and is never part of the
payload, so nothing round-trips differently.
**One thing worth naming, since it is adjacent and I would rather raise it
than have you find it:** this does change the coder of the error output for a
*pipeline update* of a running streaming job. I do not think that can bite
anyone in practice, because the old declared shape was `{failed_row:
{failed_row: …, error_message: …}, error_message: …}` while every element the
transform emits is `{failed_row: …, error_message: …}` — so a job that actually
consumed that output would already have been failing. But that is reasoning,
not something I have run, so treat it as a flag rather than a clearance.
Happy to add a `KafkaSchemaTransformTranslationTest` case pinning the
config-row round trip if you would like the guarantee written down rather than
argued.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
