Kotlin Iterable не поддерживается в Apache Beam?

Question

Луч Apache, похоже, отказывается распознавать 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 на самом деле является только интерфейсоми эта базовая коллекция все еще изменчива.Однако попытка заменить Iterable на MutableIterable продолжает вызывать ошибку.

Обновление 2 :

Я развернул задание Kotlin Dataflow, используя MutableListв соответствии с вышеизложенным, и работа не удалась с:

java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.ClassCastException:
org.apache.beam.runners.dataflow.worker.util.BatchGroupAlsoByWindowViaIteratorsFn$WindowReiterable cannot be cast to java.util.List
    at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:184)
    at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner$1.outputWindowedValue(GroupAlsoByWindowFnRunner.java:102)

Мне пришлось переключиться обратно на использование java.lang.Iterable.

Answer 1

Это похоже на ошибку в Beam Kotlin SDK.Анализ отражений для вашего @ProcessElement метода работает неправильно.Вероятно, вы можете обойти это, используя ProcessContext ctx вместо параметра @Element.

Answer 2

Я не очень знаком с kotlin, но кажется, что вам нужно импортировать import java.lang.Iterable перед использованием его в вашем коде.