Flink拓扑单元测试：MiniCluster序列化与类型推断问题咨询

我有一个包含多个Map和FlatMap转换操作的Flink拓扑，数据源与输出端均对接Kafka，Kafka记录类型为他人定义的Envelope类，该类未标记为可序列化。我希望对该拓扑进行单元测试，于是定义了一个返回Envelope列表的简单SourceFunction作为数据源：

public class MySource extends RichParallelSourceFunction<Envelope> {

    private List<Envelope> input;

    public MySource(List<Envelope> input) {
        this.input = input;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

    }

    @Override
    public void run(SourceContext<Envelope> ctx) throws Exception {
        for (Envelope listElement : inputOfSubtask) {
            ctx.collect(listElement);
        }
    }

    @Override
    public void cancel() {}
}

使用MiniClusterWithClientResource进行拓扑单元测试时，遇到三个问题：

1. Flink要求MySource必须可序列化，我将input设为transient作为临时解决方法，代码可正常编译；
2. 运行时出现如下错误：

org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'Custom Source' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.

此前使用KafkaConsumer对接Kafka集群时未出现该错误，通过添加.returns(TypeInformation.of(Envelope.class))指定类型信息可解决此问题；
3. 反序列化完成后，input被设置为null（显然是因为未定义反序列化方法）。

现提出以下问题：

1. 为何会出现上述InvalidTypesException异常？
2. MySource为何需要被序列化/反序列化？使用MiniClusterWithClientResource时能否避免该操作？
3. 我可以在MySource中编写writeObject()和readObject()方法来处理序列化，但更希望避免此方式。是否可以使用类似KeyedOneInputStreamOperatorTestHarness的框架/类来测试拓扑，无需提供可序列化的Source和Sink，从而规避初始的序列化/反序列化步骤？

1. 为什么会触发InvalidTypesException？ #

Java的泛型类型擦除会在编译阶段移除泛型参数信息，Flink自动类型推断依赖这些泛型信息来确定算子的输出类型。

官方提供的KafkaConsumer内部已经通过实现ResultTypeQueryable接口或显式指定类型信息，绕过了类型擦除的限制；而你自定义的MySource继承自RichParallelSourceFunction<Envelope>，泛型参数Envelope在运行时被擦除，Flink无法自动推断出返回类型，因此抛出该异常。

2. MySource为什么需要序列化？能不能避免？ #

Flink本质是分布式计算框架，即使使用MiniClusterWithClientResource，底层也是模拟分布式运行流程：客户端会把包含SourceFunction的作业图序列化后，传递给MiniCluster的TaskManager进程执行。所以MySource必须支持序列化才能完成跨进程传递。

这种序列化是Flink作业提交机制的核心环节，用MiniCluster时无法完全避免，但可以调整Source的实现方式来减少影响：比如不在构造函数中传入input，而是在open方法里通过配置参数、本地文件或内存缓存加载测试数据，避免把input作为需要序列化的成员变量。

3. 能否用测试框架规避序列化要求？ #

可以使用Flink官方提供的OneInputStreamOperatorTestHarness（非键控流）或KeyedOneInputStreamOperatorTestHarness（键控流）来单独测试Map/FlatMap这类算子，无需构建完整拓扑和可序列化的Source/Sink。

这种方式的核心是让测试完全运行在同一个JVM进程内，绕开Flink的分布式序列化机制，示例代码大致如下：

// 假设要测试的FlatMap算子是MyFlatMap
MyFlatMap flatMap = new MyFlatMap();
OneInputStreamOperatorTestHarness<Envelope, OutputType> testHarness = 
    new OneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(flatMap));

// 初始化测试环境
testHarness.open();

// 手动输入测试数据
testHarness.processElement(new StreamRecord<>(new Envelope(...)));

// 获取输出结果并验证
List<StreamRecord<OutputType>> outputs = testHarness.extractOutputStreamRecords();
// 编写断言逻辑验证结果正确性

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