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

[loserwang1024]

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));
    }
} 

（ps: the modify of StarRocksUtils just apply spotless)

[loserwang1024]

@PatrickRen , CC

[pvary]

Good catch @loserwang1024!

Could you please add a test case to prevent later code changes to revert this fix?

[loserwang1024]

Could you please add a test case to prevent later code changes to revert this fix?

I'd like to, but it seems no pipeline sink which is WithPreWriteTopology but not TwoPhaseCommittingSink now unless i mock one.

[pvary]

Could you please add a test case to prevent later code changes to revert this fix?

I'd like to, but it seems no pipeline sink which is WithPreWriteTopology but not TwoPhaseCommittingSink now unless i mock one.

Could you define your own one in the test itself?
Then you have free hands what it does, and does not...

[loserwang1024]

Could you define your own one in the test itself? Then you have free hands what it does, and does not...

Done it.

[pvary]

nit: Maybe some more descriptive content, like

String uid = "Uid set by the addPreWriteTopology topology";

[loserwang1024]

I've tried it before, but it shows that :
java.lang.IllegalArgumentException: Node hash must be a 32 character String that describes a hex code.

[loserwang1024]

@PatrickRen , CC, Would you like to help me review this PR?