apache / apache/beam

Ensure all RunnerHarnesses provide a valid RunnerApi.IsBounded value on all PCollections

Open
#19,410 0 comments 0 reactions 0 assignees View on GitHub
core new feature P3 portability runners
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

Fixing this requires updating 4 locations.
* Dataflow RunnerHarness
* FNAPDoFnRunner
* UnifiedWorker
* Shared libraries for this proto generation, which should cover OSS runners
**** Remove the workaround in ProcessBundleHandler.java which will assume that all PCollections are bounded, if not set.

See PCollectionTranslation.fromProto which should be always passed a valid value and not default to error or assume the PCollection is bounded.

 

Context

\===

When I was updating the java SDK to conditionally serialize some elements to reported a sampled byte size metric, I encountered this.

 
Its due to to the refactoring in my [PR/8416](https://github.com/apache/beam/pull/8416), the RehydratedComponents was pulled up a level, and shared now among all the calls to createRunnerForPTransform in the various PtransfomRunnerFactories.
 
I is now triggering some code paths which were not previously triggered for all types of PTransforms/PCollections, causing this error to occur.
 
 jsonPayload: {
  exception:  "org.apache.beam.vendor.guava.v20_0.com.google.common.util.concurrent.UncheckedExecutionException: java.lang.IllegalArgumentException: Cannot convert unknown org.apache.beam.model.pipeline.v1.RunnerApi.IsBounded to org.apache.beam.sdk.values.PCollection.IsBounded: UNSPECIFIED
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2214)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache.get(LocalCache.java:4053)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4057)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4986)
at org.apache.beam.runners.core.construction.RehydratedComponents.getPCollection(RehydratedComponents.java:144)
at org.apache.beam.fn.harness.data.PCollectionConsumerRegistry.getMultiplexingConsumer(PCollectionConsumerRegistry.java:145)
at org.apache.beam.fn.harness.DoFnPTransformRunnerFactory$Context.(DoFnPTransformRunnerFactory.java:284)
at org.apache.beam.fn.harness.DoFnPTransformRunnerFactory.createRunnerForPTransform(DoFnPTransformRunnerFactory.java:97)
at org.apache.beam.fn.harness.DoFnPTransformRunnerFactory.createRunnerForPTransform(DoFnPTransformRunnerFactory.java:63)
at org.apache.beam.fn.harness.control.ProcessBundleHandler.createRunnerAndConsumersForPTransformRecursively(ProcessBundleHandler.java:198)
at org.apache.beam.fn.harness.control.ProcessBundleHandler.createRunnerAndConsumersForPTransformRecursively(ProcessBundleHandler.java:166)
at org.apache.beam.fn.harness.control.ProcessBundleHandler.createRunnerAndConsumersForPTransformRecursively(ProcessBundleHandler.java:166)
at org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:306)
at org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:160)
at org.apache.beam.fn.harness.control.BeamFnControlClient.lambda$processInstructionRequests$0(BeamFnControlClient.java:144)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Cannot convert unknown org.apache.beam.model.pipeline.v1.RunnerApi.IsBounded to org.apache.beam.sdk.values.PCollection.IsBounded: UNSPECIFIED
at org.apache.beam.runners.core.construction.PCollectionTranslation.fromProto(PCollectionTranslation.java:88)
at org.apache.beam.runners.core.construction.PCollectionTranslation.fromProto(PCollectionTranslation.java:56)
at org.apache.beam.runners.core.construction.RehydratedComponents$3.load(RehydratedComponents.java:103)
at org.apache.beam.runners.core.construction.RehydratedComponents$3.load(RehydratedComponents.java:93)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3628)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2336)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2295)
at org.apache.beam.vendor.guava.v20_0.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2208)
... 17 more
"  
  job:  "2019-05-29_03_31_14-4799355109250203557"  
  logger:  "org.apache.beam.fn.harness.control.BeamFnControlClient"  
  *message:  "Exception while trying to handle InstructionRequest -28"*  
  portability_worker_id:  "1"  
  thread:  "16"  
  worker:  "testpipeline-pabloem-0529-05290331-75o8-harness-htz8"  
 }
 
The root of the issue is that the ProcessBundleDescriptors are invalid. The RunnerHarnesses are not setting the org.apache.beam.model.pipeline.v1.RunnerApi.IsBounded which breaks the specification and leads to this error.
 
 

Imported from Jira [BEAM-7452](https://issues.apache.org/jira/browse/BEAM-7452). Original Jira may contain additional context.
Reported by: ajamato@google.com.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.