[FLINK-36546] Handle batch sources in DataSinkTranslator

[morozov]

There's no public API in Flink to detect the boundedness of a stream, so this patch duplicates the code from StreamGraphGenerator that Flink itself uses to instantiate CommitterOperatorFactory.

[lvyanquan]

I think a test for this is still necessary.
We have two bound sources(values, MySQL when setting scan.startup.mode to snapshot), and I don't get the point that you said "the sink never receives the end-of-input signal", can you provide a more detailed description? we can add some logs in flush method if endOfInput is true to verify this.

[morozov]

@lvyanquan please see the temporary reproducer in the second commit.

and I don't get the point that you said "the sink never receives the end-of-input signal", can you provide a more detailed description?

I got confused. The problem is not that the sink never receives the end-of-input signal but that the two-phase committing sink doesn't commit the last checkpoint.

In order to reproduce the issue I had to make the following temporary changes:

1. Make ValuesSink a TwoPhaseCommittingSink.
2. Bypass the usage of reflection in DataSinkTranslator. Otherwise, the test would fail with an IllegalAccessException, which I didn't know how to address.

The point is that with the fix from the first commit, when DataSinkTranslatorBatchModeIT runs, it produces the following log message:

flink-cdc/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/sink/ValuesDataSink.java

Line 123 in 2274b7e

LOG.info("Find me in the logs. Committing {} committables.", committables.size());

4890 [PostPartition -> Sink Writer: Value Sink -> Sink Committer: Value Sink (1/1)#0] INFO org.apache.flink.cdc.connectors.values.sink.ValuesDataSink$ValuesSink - Find me in the logs. Committing 1 committables.

If the first commit is reverted, this message won't be produced because the sink won't commit.