PDGGK opened a new pull request, #39760:
URL: https://github.com/apache/beam/pull/39760
**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:
```java
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:
```java
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.
--
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]
