使用Flink1.17与flink-connector-kinesis-4.1.0遇SequenceNumber泛型异常求方案

问题原因 #

触发的异常为：

java.lang.UnsupportedOperationException: Generic types have been disabled in the ExecutionConfig and type org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber is treated as a generic type.

这是因为SequenceNumber类未被Flink类型系统正确识别，被标记为泛型类型，而当前Flink作业的ExecutionConfig默认禁用了泛型类型支持。FLINK-24943修复本应解决这个类型识别问题，但你使用的flink-connector-kinesis-4.1.0-1.17版本未包含该修复。

替代方案 #

方案1：手动注册SequenceNumber类型信息 #

在作业初始化阶段，显式将SequenceNumber注册到Flink的类型系统中，让Flink能正确识别它的类型，而非当作泛型处理：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.java.typeutils.TypeInformation;
import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber;

public class KinesisJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 注册SequenceNumber的类型信息
        env.getConfig().registerTypeInformation(SequenceNumber.class, TypeInformation.of(SequenceNumber.class));
        
        // 后续的Kinesis源/连接器配置和作业逻辑
        // ...
        
        env.execute("Kinesis Streaming Job");
    }
}

方案2：显式指定状态的序列化器（针对自定义状态场景） #

如果你的作业中直接使用了包含SequenceNumber的状态（比如自定义Checkpoint状态），可以为状态描述符显式指定类型信息，避免泛型识别问题：

import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.typeutils.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeHint;
import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber;
import org.apache.flink.api.java.tuple.Tuple2;

// 定义包含SequenceNumber的状态描述符
ListStateDescriptor<Tuple2<String, SequenceNumber>> stateDesc = new ListStateDescriptor<>(
    "kinesis-offset-state",
    TypeInformation.of(new TypeHint<Tuple2<String, SequenceNumber>>() {})
);

方案3：临时启用泛型类型支持（不推荐，仅应急） #

如果上述方案无法快速实施，可以临时开启Flink的泛型类型支持，但这会关闭类型安全检查，可能带来性能或序列化问题，仅作为临时应急方案：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().enableGenericTypes();

方案4：升级到包含修复的版本 #

如果允许升级，建议直接使用包含FLINK-24943修复的连接器版本，比如flink-connector-kinesis-4.2.0-1.17或更高的兼容Flink 1.17的版本，从根源解决问题。

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