schizophrenicmaniac commented on code in PR #40481:
URL: https://github.com/apache/beam/pull/40481#discussion_r4281877745
##########
sdks/java/io/splunk/src/main/java/org/apache/beam/sdk/io/splunk/SplunkIO.java:
##########
@@ -166,7 +166,8 @@ public PCollection<SplunkWriteError>
expand(PCollection<SplunkEvent> input) {
// Return a PCollection<SplunkWriteError>
return input
.apply("Create KV pairs", CreateKeys.of(parallelism()))
- .apply("Write Splunk events", ParDo.of(writer));
+ .apply("Write Splunk events", ParDo.of(writer))
+ .setCoder(SplunkWriteErrorCoder.of());
Review Comment:
Yes, that works. I dropped `SplunkWriteErrorCoder` and added a
`SplunkWriteErrorSchemaProvider` with hand-written to/from row functions
(fields `statusCode`, `statusMessage`, `payload`). `SplunkWriteError` now uses
it via `@DefaultSchema` instead of `AutoValueSchema`, so the `setCoder` call in
`SplunkIO.Write` is gone and the output is inferred as a `SchemaCoder` like
before, just with actual fields now. The schema transform still builds its own
error rows since it adds `failed_row`.
I didn't touch the get-prefix handling in `AutoValueSchema`. That would
change the inferred schema, and so the coder, of any other AutoValue type that
has non-get accessors, so it seemed better kept out of this PR.
--
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]
