Unit test Flink job using DeltaSource & DeltaSink

[Timelessprod]

Hello,

I'm quite new to Flink and I try to build a simple project which reads Delta tables, do some processing using Flink and then stores the output to (other) Delta tables. To make things clean I want to unit test all (or most) of the function I write, including the Flink jobs.

As I use Delta, sourcea of my jobs are DeltaSources and sinks are DeltaSinks. However to run unit tests I need to be able to inject custom data in the jobs and check the output to see if it matches what I expect. As the documentation doesn't state how to achieve that either on the Flink documentation and Delta documentation, I tried to do things myself. I defined the sink below to collect the job outputs and use the fromElements() method as a source to inject data. The problem I have now is that DeltaSink doesn't extends or implements the Flink Sink interface thus I cannot swap a DeltaSink for a CollectSink in my jobs without getting typing errors. Same goes for the DeltaSource when I try to use fromElements().

Is there a proper way to do unit test when using the Delta connector with Flink ? If so where am I supposed to find documentation as it is not written in this repo nor on delta.io website ?

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class CollectSink<T> implements Sink<T> {
    private final List<T> elements = new CopyOnWriteArrayList<>();

    @Override
    public SinkWriter<T> createWriter(InitContext initContext) throws IOException {
        return new SinkWriter<T>() {
            @Override
            public void write(T element, SinkWriter.Context context) {
                elements.add(element);
            }

            @Override
            public void flush(boolean b) throws IOException, InterruptedException {
                elements.clear();
            }

            @Override
            public void writeWatermark(Watermark watermark) throws IOException, InterruptedException {
                SinkWriter.super.writeWatermark(watermark);
            }

            @Override
            public void close() {
            }
        };
    }

    public List<T> getElements() {
        return elements;
    }
}

Thank you.

PS: Any up-to-date (Flink 1.20 ideally or 1.19) documentation and/or sample project about Flink and/or Delta Flink Connector is welcome :) Those are very hard to find online which makes starting a project difficult.