@VisibleForTesting
    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 {
            stream.transform(
                    SINK_WRITER_PREFIX + sinkName,
                    CommittableMessageTypeInfo.noOutput(),
                    new DataSinkWriterOperatorFactory<>(sink, schemaOperatorID));
        }
    }
@Test
    void testPreWriteWithoutCommitSink() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ArrayList<Event> mockEvents = Lists.newArrayList(new EmptyEvent(), new EmptyEvent());
        DataStreamSource<Event> inputStream = env.fromCollection(mockEvents);
        DataSinkTranslator translator = new DataSinkTranslator();

        // Node hash must be a 32 character String that describes a hex code
        String uid = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        MockPreWriteWithoutCommitSink mockPreWriteWithoutCommitSink =
                new MockPreWriteWithoutCommitSink(uid);
        translator.sinkTo(
                inputStream,
                mockPreWriteWithoutCommitSink,
                "testPreWriteWithoutCommitSink",
                new OperatorID());

        // Check if the `addPreWriteTopology` is called, and the uid is set when the transformation
        // added
        OneInputTransformation<Event, Event> oneInputTransformation =
                (OneInputTransformation) env.getTransformations().get(0);
        Transformation<?> reblanceTransformation = oneInputTransformation.getInputs().get(0);
        Assertions.assertEquals(uid, reblanceTransformation.getUserProvidedNodeHash());
    }