VS Code+Maven环境下Flink访问Java POJO失败，报InaccessibleObjectException

环境信息 #

OS : Windows 11
JDK : open-jdk 17.0.2
maven : apache-maven 3.9.6
apache flink : 1.18.1 (安装在WSL 2 ubuntu上)
visual studio code : x64 1.87.2

问题代码 #

Car.java（自定义POJO类） #

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class Car{

    private String brand;
    private int price;
}

Flink MongoDB Sink主代码 #

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

List<Car> cars = new ArrayList<>();
cars.add(new Car("BMW", 500));
cars.add(new Car("Kia", 300));
cars.add(new Car("Ford", 600));

DataStream<Car> stream = env.fromCollection(cars);

MongoSink<Car> sink = MongoSink.<Car>builder()
            .setUri("mongodb://127.0.0.1:27017")
            .setDatabase("class_db")
            .setCollection("class_coll")
            .setBatchSize(1000)
            .setBatchIntervalMs(1000)
            .setMaxRetries(3)
            .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
            .setSerializationSchema(
                    (input, context) 
                        -> {
                            Document doc = new Document(input.getBrand(), input.getPrice());
                            System.out.println(doc);
                            return new InsertOneModel<>(BsonDocument.parse(doc.toJson()));
                        })
            .build();

stream.sinkTo(sink);

env.execute("MongoDB POJO Test");
env.close();

抛出异常 #

Exception in thread "main" java.io.UncheckedIOException: java.io.IOException: Serializing the source elements failed: java.lang.reflect.InaccessibleObjectException: Unable to make field private final java.lang.Object[] java.util.Arrays$ArrayList.a accessible: module java.base does not "opens java.util" to unnamed module @2a32de6c
        at org.apache.flink.streaming.api.functions.source.FromElementsFunction.setOutputType(FromElementsFunction.java:162)  
        at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySetOutputType(StreamingFunctionUtils.java:84)  
        at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.setOutputType(StreamingFunctionUtils.java:60)     
        at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.setOutputType(AbstractUdfStreamOperator.java:146)
        at org.apache.flink.streaming.api.operators.SimpleOperatorFactory.setOutputType(SimpleOperatorFactory.java:118)       
        at org.apache.flink.streaming.api.graph.StreamGraph.addOperator(StreamGraph.java:434)
        at org.apache.flink.streaming.api.graph.StreamGraph.addOperator(StreamGraph.java:402)
        at org.apache.flink.streaming.api.graph.StreamGraph.addLegacySource(StreamGraph.java:356)
        at org.apache.flink.streaming.runtime.translators.LegacySourceTransformationTranslator.translateInternal(LegacySourceTransformationTranslator.java:66)
        at org.apache.flink.streaming.runtime.translators.LegacySourceTransformationTranslator.translateForStreamingInternal(LegacySourceTransformationTranslator.java:53)
        at org.apache.flink.streaming.runtime.translators.LegacySourceTransformationTranslator.translateForStreamingInternal(LegacySourceTransformationTranslator.java:40)
        at org.apache.flink.streaming.api.graph.SimpleTransformationTranslator.translateForStreaming(SimpleTransformationTranslator.java:62)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.translate(StreamGraphGenerator.java:860)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.transform(StreamGraphGenerator.java:590)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.getParentInputIds(StreamGraphGenerator.java:881)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.translate(StreamGraphGenerator.java:839)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.transform(StreamGraphGenerator.java:590)
        at org.apache.flink.streaming.api.graph.StreamGraphGenerator.generate(StreamGraphGenerator.java:328)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2289)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2280)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2266)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:2093)
        at com.aaa.test.FlinkMongoClassTest.main(FlinkMongoClassTest.java:49)
Caused by: java.io.IOException: Serializing the source elements failed: java.lang.reflect.InaccessibleObjectException: Unable to make field private final java.lang.Object[] java.util.Arrays$ArrayList.a accessible: module java.base does not "opens java.util" to unnamed module @2a32de6c
        at org.apache.flink.streaming.api.functions.source.FromElementsFunction.serializeElements(FromElementsFunction.java:139)
        at org.apache.flink.streaming.api.functions.source.FromElementsFunction.setOutputType(FromElementsFunction.java:160)  
        ... 22 more
Caused by: java.lang.RuntimeException: java.lang.reflect.InaccessibleObjectException: Unable to make field private final java.lang.Object[] java.util.Arrays$ArrayList.a accessible: module java.base does not "opens java.util" to unnamed module @2a32de6c
        at com.twitter.chill.java.ArraysAsListSerializer.<init>(ArraysAsListSerializer.java:69)
        at org.apache.flink.api.java.typeutils.runtime.kryo.FlinkChillPackageRegistrar.registerSerializers(FlinkChillPackageRegistrar.java:67)
        at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.getKryoInstance(KryoSerializer.java:513)
        at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.checkKryoInitialized(KryoSerializer.java:522)      
        at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:348)
        at org.apache.flink.streaming.api.functions.source.FromElementsFunction.serializeElements(FromElementsFunction.java:136)
        ... 23 more
Caused by: java.lang.reflect.InaccessibleObjectException: Unable to make field private final java.lang.Object[] java.util.Arrays$ArrayList.a accessible: module java.base does not "opens java.util" to unnamed module @2a32de6c
        at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:354)
        at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:297)

临时解决方法 #

将输入流类型替换为Flink原生Tuple2后，代码可正常运行：

List<Tuple2<String, Integer>> cars = new ArrayList<>();
cars.add(new Tuple2<>("BMW", 500));
cars.add(new Tuple2<>("Kia", 300));
cars.add(new Tuple2<>("Ford", 600));

DataStream<Tuple2<String, Integer>> stream = env.fromCollection(cars);

问题根源与解决方案 #

根源分析 #

异常核心是Java 17的模块系统限制：Flink默认使用的Kryo序列化库尝试反射访问java.util.Arrays$ArrayList的私有字段，但Java 17的模块未将java.util开放给未命名模块，导致反射失败。而Tuple2能正常运行是因为Flink对原生Tuple类型有内置的序列化支持，无需依赖Kryo的反射逻辑。

解决方案 #

方案1：添加JVM启动参数开放模块权限

在运行配置中添加以下JVM参数，允许未命名模块访问java.util包：

--add-opens java.base/java.util=ALL-UNNAMED

如果通过Maven运行，可在pom.xml的maven-surefire-plugin中配置：

<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-surefire-plugin</artifactId>
    <version>3.2.5</version>
    <configuration>
        <argLine>--add-opens java.base/java.util=ALL-UNNAMED</argLine>
    </configuration>
</plugin>

方案2：让Car类符合Flink的POJO标准

Flink对POJO类有特定要求，满足后会使用专门的POJO序列化器，而非Kryo：

1. 类必须是public的
2. 必须有无参构造函数（Lombok的@Data默认生成无参构造，但同时使用@AllArgsConstructor会覆盖，需显式添加@NoArgsConstructor）
3. 所有字段必须是public，或者有对应的getter/setter方法（Lombok的@Data已满足）

修改后的Car.java：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Car{

    private String brand;
    private int price;
}

方案3：显式指定使用Flink的POJO类型信息

在创建DataStream时，显式告诉Flink这是POJO类型：

DataStream<Car> stream = env.fromCollection(cars, TypeInformation.of(Car.class));

或者使用PojoTypeInformation：

PojoTypeInformation<Car> carTypeInfo = PojoTypeInformation.of(Car.class);
DataStream<Car> stream = env.fromCollection(cars, carTypeInfo);

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