[
https://issues.apache.org/jira/browse/BEAM-5674?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Henning Rohde reassigned BEAM-5674:
-----------------------------------
Assignee: Sam Rohde (was: Henning Rohde)
> DataflowRunner.Deduplicate/WithKeys cannot infer Coder for K when running
> with experiment "enable_custom_pubsub_source"
> -----------------------------------------------------------------------------------------------------------------------
>
> Key: BEAM-5674
> URL: https://issues.apache.org/jira/browse/BEAM-5674
> Project: Beam
> Issue Type: Bug
> Components: runner-dataflow
> Affects Versions: 2.7.0
> Environment: OS: Linux 64bit,
> Platform: Dataflow
> Reporter: Duy Le
> Assignee: Sam Rohde
> Priority: Major
> Labels: easyfix, patch
>
> *Bug*: DataflowRunner.Deduplicate/WithKeys cannot infer Coder for K when
> running with experiment "enable_custom_pubsub_source"
> *Steps*
> # Start a Beam pipeline with DataflowRunner using
> ExperimentalOptions."enable_custom_pubsub_source"
> # Observe the result when the pipeline is being constructed.
> *Actual*: An error thrown
> {color:#FF0000}Unable to return a default Coder for
> PubsubIO.Read/PubsubUnboundedSource/Read(PubsubSource)/DataflowRunner.Deduplicate/WithKeys/AddKeys/Map/ParMultiDo(Anonymous).output
> [PCollection]. Correct one of the following root causes:{color}
> {color:#FF0000} No Coder has been manually specified; you may do so using
> .setCoder().{color}
> {color:#FF0000} Inferring a Coder from the CoderRegistry failed: Cannot
> provide coder for parameterized type org.apache.beam.sdk.values.KV<K, V>:
> Unable to provide a Coder for K.{color}
> {color:#FF0000} Building a Coder using a registered CoderProvider
> failed.{color}
> *Expected*: The pipeline should be constructed successfully.
> *Root cause*:
> In
> {noformat}
> DataflowRunner.Deduplicate{noformat}
> transform, it applies
> {code:java}
> WithKeys.of(){code}
> transform to an input of *ValueWithRecordId* with a function in the style of
> Java 8 lambda.
>
>
> As the Javadoc states that:
>
> {code:java}
> If using a lambda in Java 8, {@link #withKeyType(TypeDescriptor)} must be
> called on the result {@link PTransform}{code}
>
> *Suggested Solution*:
> Since the lambda function returns a hashed code (_int_) of the
> {code:java}
> value.getId(){code}
> (_byte[]_), can we just use
> {code:java}
> withKeyType(TypeDescriptors.integers()){code}
> right after the
> {code:java}
> WithKeys.of(){code}
> method?
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
