Kotlin环境下Apache Beam 2.27.0使用FlatMapElements报非序列化Lambda异常

Kotlin编译器默认不会为生成的Lambda类自动实现Serializable接口，和Java编译器的行为差异是触发异常的核心原因。Java环境下，当Lambda的目标类型为实现了Serializable的函数式接口时，javac会自动生成可序列化的Lambda代理类；而Kotlin默认仅在目标类型明确标记为可序列化、且无额外配置时，才会生成对应的序列化实现，导致Beam运行时在序列化封装在FlatMapElements内部的DoFn时，识别到Lambda为非可序列化对象，抛出NotSerializableException。

方案1：全局对齐Java编译行为，开启Lambda序列化配置 #

在pom.xml的kotlin-maven-plugin配置段增加编译参数，让所有Kotlin生成的Lambda默认实现Serializable接口，完全对齐Java环境的编译行为：

<plugin>
  <groupId>org.jetbrains.kotlin</groupId>
  <artifactId>kotlin-maven-plugin</artifactId>
  <!-- 原有版本、kapt、源码路径等配置保持不变 -->
  <configuration>
    <compilerArgs>
      <!-- 新增该参数，强制所有Lambda实现Serializable -->
      <arg>-Xjava-lambda-serialization=always</arg>
    </compilerArgs>
  </configuration>
</plugin>

配置生效后，原有使用MapElements/FlatMapElements搭配Lambda的业务代码无需任何修改即可正常运行，完全匹配原来Java环境的开发体验。

注意：该参数从Kotlin 1.5.0开始正式支持，与当前项目使用的Kotlin版本完全适配，不会触发1.4.30版本的类读取兼容问题。

方案2：显式声明Lambda的可序列化类型 #

如果不希望修改全局编译配置，可在使用Lambda时显式将其标记为Beam的SerializableFunction类型，强制编译器为该Lambda生成可序列化实现：

// 原有错误写法（无明确类型标记，生成的Lambda不可序列化）
pCollection.apply(
  FlatMapElements
    .into(TypeDescriptor.of(MyObject::class.java))
    .via { kv: KV<String, String> ->
      listOf(OBJECT_MAPPER.readValue(kv.value, MyObject::class.java))
    }
)

// 修正写法（明确指定目标类型为SerializableFunction）
pCollection.apply(
  FlatMapElements
    .into(TypeDescriptor.of(MyObject::class.java))
    .via(
      { kv: KV<String, String> ->
        listOf(OBJECT_MAPPER.readValue(kv.value, MyObject::class.java))
      } as SerializableFunction<KV<String, String>, Iterable<MyObject>>
    )
)

也可直接定义类型明确的可序列化函数变量，传递给转换组件：

val mapFunction: SerializableFunction<KV<String, String>, Iterable<MyObject>> = { input ->
  listOf(OBJECT_MAPPER.readValue(input.value, MyObject::class.java))
}

pCollection.apply(
  FlatMapElements
    .into(TypeDescriptor.of(MyObject::class.java))
    .via(mapFunction)
)

• 确保Lambda捕获的外部引用可被序列化：以业务中使用的Jackson ObjectMapper为例，该类本身未实现Serializable接口，需将其定义为companion object中的静态常量，避免被作为类实例成员参与序列化流程，否则哪怕Lambda本身实现了序列化，也会因为捕获的对象不可序列化抛出同类异常。
• 不要通过类型别名隐式传递可序列化函数：Kotlin的typealias仅为类型别名，不会触发编译器为Lambda生成序列化实现，必须显式指定目标函数类型为SerializableFunction。
• 避免在Lambda中捕获可变的类实例成员，所有跨Lambda引用的常量、工具类都应放在伴生对象中定义，或标记为@Transient。

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