如何实现可配合Flink .sinkTo使用的测试Sink？

你可以自己实现一个适配Flink .sinkTo 方法的TestSink，核心是实现Flink的Sink<T>接口，内部用线程安全的集合存储输出数据，方便后续断言验证。

自定义TestSink实现 #

import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.core.io.SimpleVersionedSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

public class TestSink<T> implements Sink<T> {

    // 用线程安全队列存储收集到的元素，适配多并行子任务写入场景
    private static final ConcurrentLinkedQueue<Object> collectedElements = new ConcurrentLinkedQueue<>();

    public static <T> TestSink<T> getInstance() {
        return new TestSink<>();
    }

    // 获取收集到的所有元素，返回新列表避免并发修改问题
    public static List<Object> getCollectedElements() {
        return new ArrayList<>(collectedElements);
    }

    // 测试前清空数据，避免不同测试用例互相干扰
    public static void clearCollectedElements() {
        collectedElements.clear();
    }

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

            @Override
            public void flush(boolean endOfInput) {}

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

    @Override
    public SimpleVersionedSerializer<String> getCommittableSerializer() {
        // 测试场景不需要提交机制，直接返回null即可
        return null;
    }
}

测试中的使用示例 #

import org.junit.Test;
import static org.junit.Assert.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JobTest {

    @Test
    public void testJobOutput() throws Exception {
        // 测试前清空历史数据
        TestSink.clearCollectedElements();

        // 构建Flink作业，替换原KafkaSink为TestSink
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 如果是流作业，可转为批处理模式方便测试
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
        env.fromElements("order_001", "order_002", "order_003")
           .map(String::toUpperCase)
           .sinkTo(TestSink.getInstance());

        // 执行作业
        env.execute("Test Job");

        // 获取收集到的数据并执行断言
        List<Object> results = TestSink.getCollectedElements();
        assertEquals(3, results.size());
        assertTrue(results.contains("ORDER_001"));
        assertTrue(results.contains("ORDER_003"));
    }
}

注意事项 #

• 线程安全：使用ConcurrentLinkedQueue确保多并行子任务写入时不会出现并发问题
• 场景适配：如果是流作业测试，建议转为批处理模式，或者手动触发水位线、等待作业处理完成，避免因流的无限性导致断言时数据未完全写入
• 类型支持：如果处理POJO或自定义类型，需确保元素重写了equals方法，否则断言比较会失效

内容的提问来源于stack exchange，提问作者user3497321