[FLINK-35149] Fix DataSinkTranslator#sinkTo ignoring pre-write topology if not TwoPhaseCommittingSink

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: cdc-3.1.0
• Fix Version/s: cdc-3.2.0
• Component/s: Flink CDC
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Hongshun Wang
• Created: 18/Apr/24 02:14
• Updated: 1 week ago 09:33

Description

Current , when sink is not instanceof TwoPhaseCommittingSink, use input.transform rather than stream. It means that pre-write topology will be ignored.

private void sinkTo(
        DataStream<Event> input,
        Sink<Event> sink,
        String sinkName,
        OperatorID schemaOperatorID) {
    DataStream<Event> stream = input;
    // Pre write topology
    if (sink instanceof WithPreWriteTopology) {
        stream = ((WithPreWriteTopology<Event>) sink).addPreWriteTopology(stream);
    }

    if (sink instanceof TwoPhaseCommittingSink) {
        addCommittingTopology(sink, stream, sinkName, schemaOperatorID);
    } else {
        input.transform(
                SINK_WRITER_PREFIX + sinkName,
                CommittableMessageTypeInfo.noOutput(),
                new DataSinkWriterOperatorFactory<>(sink, schemaOperatorID));
    }
} 

Issue Links

• links to: GitHub Pull Request #3233