我收到错误：使用 Kotlin 时 Apache Beam 中的 MapElements 转换出现

Question

我正在探索Github上的 GoogleCloudPlatform 提供的 Apache Beam 数据流模板。特别是，我正在将PubSubToBigQuery模板从 Java 转换为 Kotlin。通过这样做，我在在线转换中得到了一个Overload ambiguity resolution error 。错误信息是：MapElements.input(...).via(...)274Error:(62, 22) Kotlin: Cannot choose among the following candidates without completing type inference: public final fun <NewInputT : Any!> via(fn: ((input: BigQueryInsertError!) -> FailsafeElement<String!, String!>!)!): MapElements<BigQueryInsertError!, FailsafeElement<String!, String!>!>! defined in org.apache.beam.sdk.transforms.MapElementspublic final fun <NewInputT : Any!> via(fn: ((input: BigQueryInsertError!) -> FailsafeElement<String!, String!>!)!): MapElements<BigQueryInsertError!, FailsafeElement<String!, String!>!>! defined in org.apache.beam.sdk.transforms.MapElements相关的 Java 代码片段是：/*     * Step 3 Contd.     * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement     */    PCollection<FailsafeElement<String, String>> failedInserts =        writeResult            .getFailedInsertsWithErr()            .apply(                "WrapInsertionErrors",                MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())                    .via((BigQueryInsertError e) -> wrapBigQueryInsertError(e)))            .setCoder(FAILSAFE_ELEMENT_CODER);Kotlin 转换如下所示：/*     * Step 3 Contd.     * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement      */val failedInserts: PCollection<FailsafeElement<String, String>> =            writeResult.failedInsertsWithErr            .apply(                "WrapInsertionErrors",                MapElements.into(FAILSAFE_ELEMENT_CODER.encodedTypeDescriptor)                    .via { e: BigQueryInsertError -> wrapBigQueryInsertError(e) })            .setCoder(FAILSAFE_ELEMENT_CODER)我不知道如何解决这个问题。你能帮忙的话，我会很高兴。

Answer 1

原因是 Java 和 Kotlin 的重载规则略有不同，这意味着在 Kotlin 中有两个匹配的重载；

public <NewInputT> MapElements<NewInputT, OutputT> via(ProcessFunction<NewInputT, OutputT> fn)
public <NewInputT> MapElements<NewInputT, OutputT> via(SerializableFunction<NewInputT, OutputT> fn)

最简单的解决方法是将 lambda 显式指定为 aSerializableFunction以获得正确的重载；

.via<BigQueryInsertError> (SerializableFunction { e: BigQueryInsertError -> wrapBigQueryInsertError(e) }))