Tag the Kafka write error output with the schema it actually emits

[PDGGK]

Please add a meaningful description for your change here

Follow-up to #39759, which fixed the identical defect in TFRecordWriteSchemaTransformProvider. Same bug, different module, so it is a separate PR.

ErrorCounterFn (and GenericRecordErrorCounterFn) are handed errorSchema = ErrorHandling.errorSchema(inputSchema) and emit ErrorHandling.errorRecord(errorSchema, row, e), so every row on ERROR_TAG carries exactly that schema. The collection was then tagged with the wrapper applied a second time:

Schema errorSchema = ErrorHandling.errorSchema(inputSchema);                    // :246
...
new ErrorCounterFn("Kafka-write-error-counter", toBytesFn, errorSchema, handleErrors)
...
receiver.get(ERROR_TAG).output(ErrorHandling.errorRecord(errorSchema, row, e)); // :161
...
outputTuple.get(ERROR_TAG).setRowSchema(ErrorHandling.errorSchema(errorSchema)); // :301

errorSchema(x) is {failed_row: Row(x), error_message: STRING}, so the declared shape becomes {failed_row: {failed_row: …, error_message: …}, error_message: …} — which no element this transform produces can match.

KafkaReadSchemaTransformProvider gets it right in the same package, and so do JavaFilterTransformProvider, JavaMapToFieldsTransformProvider, PubsubRowToMessage, PubsubWriteSchemaTransformProvider and BigQueryStorageWriteApiSchemaTransformProvider.

Why the existing tests did not catch it

KafkaWriteSchemaTransformProviderTest exercises ErrorCounterFn by applying the ParDo directly and then calling setRowSchema(errorSchema) itself:

PCollectionTuple output = input.apply(ParDo.of(new ErrorCounterFn(...)).withOutputTags(...));
output.get(ERROR_TAG).setRowSchema(errorSchema);        // the test's own, correct, tagging

So the four ErrorFn tests never reach line 301. The one test that does build the whole transform, testBuildTransformWithManaged, does not look at the output schema.

Test

One, and it needs no runner — the schema is fixed while the graph is built, so it runs in the ordinary :sdks:java:io:kafka:test task. It goes through expand(), which is the gap above.

Restoring the second wrap fails it and nothing else:

7 tests completed, 1 failed

KafkaWriteSchemaTransformProviderTest > testErrorOutputCarriesTheSchemaErrorCounterFnEmits FAILED
    java.lang.AssertionError:
    expected:<{ failed_row: ROW { bytes: BYTES }, error_message: STRING }>
     but was:<{ failed_row: ROW { failed_row: ROW { bytes: BYTES }, error_message: STRING }, error_message: STRING }>

Note this needs a clean recompile to reproduce — running it incrementally on top of a previous build makes KafkaIO.writeRecords throw NullPointerException: Null eosTriggerTimeout in both this test and testBuildTransformWithManaged, from stale AutoValue output rather than from the change. With --rerun-tasks only the one test fails.

spotlessJavaCheck, checkstyleMain and checkstyleTest on :sdks:java:io:kafka are clean.

Are there user-facing changes?

Yes, for anyone reading the errors output of the Kafka write SchemaTransform: it is now tagged with the schema its rows actually have. Code that hardcoded the doubly-nested shape to work around this would need updating, but such code could not have been reading real rows successfully.

[github-actions]

Assigning reviewers:

R: @chamikaramj for label java.
R: @sjvanrossum for label kafka.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[sjvanrossum]

Looks good to me at first glance, but I need to take a moment to figure out if we need to add support for this breaking change in KafkaIO's upgrade compatibility module too.

[PDGGK]

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:

@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.