[BEAM-13584] Add "Write.withResults()" for all Java SDK IOs

• Type: Improvement
• Status: Open
• Priority: P3
• Resolution: Unresolved
• Component/s: io-ideas
• Assignee: Unassigned
• Reporter: Alexey Romanenko
• Created: 30/Dec/21 16:08
• Updated: 04/Jun/22 22:43

Description

Currently, by default, a "Write" transform is considered as a final PTransform in a pipeline and it returns PDone that can't be used later for next pipeline stages.

Though, sometimes it's needed to use the results of write operation as input PCollection downstream of pipeline and some IOs already support this. For example, JdbcIO.Write has the methods withResults() and withWriteResults() that returns a PTransform with output PCollection<> of write results and it can be used to analyse the results or, along with Wait transform, to write into the second database only if writes into the first one were finished, like this:

PCollection<Void> firstWriteResults = data.apply(JdbcIO.write()
     .withDataSourceConfiguration(CONF_DB_1).withResults());
data.apply(Wait.on(firstWriteResults))
     .apply(JdbcIO.write().withDataSourceConfiguration(CONF_DB_2));

So, it would be useful to add a WriteWithResults transform for other IOs that can be used in the same way.

Sub-Tasks

• Add "withResults()" for KafkaIO.write: Open