如何基于新DataSource API对Flink作业进行单元测试？

针对新的Flink DataSource API测试难度高、泛型易出问题的情况，以下是几种无需搭建完整Kafka集群的作业单元测试方案：

1. 基于Flink内置测试工具+自定义收集Sink #

利用Flink提供的Sources工具类快速构造测试用Source，搭配自定义Sink收集输出结果，结合MiniCluster模拟集群环境执行作业：

示例代码 #

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.Source;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.junit.ClassRule;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

public class ExampleJobTest {

    // 启动本地MiniCluster模拟Flink集群
    @ClassRule
    public static MiniClusterWithClientResource miniCluster =
            new MiniClusterWithClientResource(new MiniClusterResourceConfiguration.Builder()
                    .setNumberSlotsPerTaskManager(1)
                    .setNumberTaskManagers(1)
                    .build());

    @Test
    public void testFullJobLogic() throws Exception {
        // 用Flink内置的DataGeneratorSource构造测试数据（替代Kafka Source）
        Source<String> testSource = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) value -> "test-data-" + value,
                10,
                new SimpleStringSchema()
        );
        
        Source<String> testOtherSource = Sources.fromElements("other-test-1", "other-test-2");

        // 自定义Sink，收集输出结果用于断言
        List<String> outputResults = new ArrayList<>();
        SinkFunction<String> testSink = new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) {
                outputResults.add(value);
            }
        };

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 测试环境设置并行度为1，便于结果验证

        // 初始化作业，传入测试用的Source和Sink
        new Example(testSource, testOtherSource, testSink).build(env);
        
        // 执行作业
        env.execute("Test Example Job");

        // 验证输出结果符合预期
        assert outputResults.contains("test-data-0");
        assert outputResults.contains("other-test-1");
    }
}

2. 依赖注入适配新API，复用测试逻辑 #

保持作业的可插拔设计，将构造参数从SourceFunction改为新的Source接口，测试时直接替换为Flink内置的测试Source实现，避免自己实现Source时的泛型问题：

调整后的作业代码（适配新API） #

public class Example {

    private final Source<String> someSource;
    private final Source<String> someOtherSource;
    private final SinkFunction<String> someSink;

    // 构造参数改为新的Source接口
    Example(
        Source<String> someSource,
        Source<String> someOtherSource,
        SinkFunction<String> someSink
    ) {
        this.someSource = someSource;
        this.someOtherSource = someOtherSource;
        this.someSink = someSink;
    }
    
    void build(StreamExecutionEnvironment env) {
        // 基于新API构建作业逻辑
        env.fromSource(someSource, WatermarkStrategy.noWatermarks(), "Test Source")
           .union(env.fromSource(someOtherSource, WatermarkStrategy.noWatermarks(), "Other Test Source"))
           // ... 业务逻辑处理 ...
           .addSink(someSink);
    }
    
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 生产环境使用KafkaSource
        Example example = new Example(
            KafkaSource.<String>builder()
                .setBootstrapServers("kafka-broker:9092")
                .setTopics("input-topic")
                .setGroupId("flink-consumer-group")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build(),
            KafkaSource.<String>builder()
                .setBootstrapServers("kafka-broker:9092")
                .setTopics("other-input-topic")
                .setGroupId("flink-consumer-group")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build(),
            KafkaSink.<String>builder()
                .setBootstrapServers("kafka-broker:9092")
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                    .setTopic("output-topic")
                    .setValueSerializationSchema(new SimpleStringSchema())
                    .build())
                .build()
        );
        
        example.build(env);
        env.execute("Production Example Job");
    }
}

测试时直接传入Sources.fromElements()、DataGeneratorSource等内置测试Source，无需自己实现复杂的泛型代码。

3. 使用嵌入式Kafka（轻量替代方案） #

如果需要更贴近生产环境的测试，可使用kafka-streams-test-utils提供的EmbeddedKafkaCluster，它是轻量的嵌入式集群，无需单独部署Kafka：

核心步骤 #

• 引入kafka-streams-test-utils依赖
• 在测试中启动嵌入式Kafka集群，创建测试Topic
• 生产测试数据到嵌入式Kafka，用真实的KafkaSource消费，KafkaSink生产后再验证结果
• 测试结束后关闭嵌入式集群

这种方案能更真实地模拟生产环境的Kafka交互，但相比纯模拟Source/Sink会稍重，适合需要验证Kafka序列化/反序列化逻辑的场景。

内容的提问来源于stack exchange，提问作者Peter C. Glade