Apache Beam不支持Kotlin Iterable？

Question

阿帕奇梁似乎拒绝承认Kotlin的Iterable。这是一个示例代码：

@ProcessElement
fun processElement(
    @Element input: KV<String, Iterable<String>>, receiver: OutputReceiver<String>
) {
    val output = input.key + "|" + input.value.toString()
    println("output: $output")
    receiver.output(output)
}

我得到以下奇怪的错误：

java.lang.IllegalArgumentException:
   ...PrintString, @ProcessElement processElement(KV, OutputReceiver), @ProcessElement processElement(KV, OutputReceiver):
   @Element argument must have type org.apache.beam.sdk.values.KV<java.lang.String, java.lang.Iterable<? extends java.lang.String>>

果然，如果我用Iterable替换java.lang.Iterable，相同的代码工作得很好。我究竟做错了什么？

依赖项版本：

• kotlin-jvm：1.3.21
• org.apache.beam：2.11.0

这是一个完整代码和堆栈跟踪的要点：

• https://gist.github.com/marcoslin/e1e19afdbacac9757f6974592cfd8d7f#file-apache-beam-iterable-notworking-kt

更新：

经过一些试验和错误后，我发现虽然List<String>抛出类似的异常，但MutableList<String>实际上有效：

class PrintString: DoFn<KV<String, MutableList<String>>, String>() {
    @ProcessElement
    fun processElement(
        @Element input: KV<String, MutableList<String>>, receiver: OutputReceiver<String>
    ) {
        val output = input.key + "|" + input.value.toString()
        println("output: $output")
        receiver.output(output)
    }
}

所以，这提醒我Kotlin的Immutable集合实际上只是界面，底层集合仍然是可变的。但是，尝试用Iterable替换MutableIterable继续引发错误。

Answer 1

我对kotlin不是很熟悉，但似乎你需要在代码中使用它之前导入import java.lang.Iterable。

Answer 2

这看起来像Beam Kotlin SDK中的一个错误。您的@ProcessElement方法的反射分析无法正常工作。你可以通过使用ProcessContext ctx而不是使用@Element参数来解决这个问题。