[jira] [Closed] (BEAM-5674) DataflowRunner.Deduplicate/WithKeys cannot infer Coder for K when running with experiment "enable_custom_pubsub_source"

["Sam Rohde (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-5674?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Sam Rohde closed BEAM-5674.
---------------------------
       Resolution: Fixed
    Fix Version/s: 2.10.0

> 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
>             Fix For: 2.10.0
>
>          Time Spent: 1h
>  Remaining Estimate: 0h
>
> *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)