Uploaded image for project: 'Beam'
  1. Beam
  2. BEAM-11028

NullPointerException when running Flink Nexmark tests on Streaming after switch to SDF based translation

Details

    Description

      When running Nexmark on Streaming mode with Flink locally via:
       
      /gradlew :sdks:java:testing:nexmark:run \
            -Pnexmark.runner=":runners:flink:1.10" \
            -Pnexmark.args=" \
                --runner=FlinkRunner \
                --streaming=true \
                --suite=SMOKE \
                --manageResources=false \
                --monitorJobs=true \
                --enforceEncodability=true \
                --enforceImmutability=true"
       
      I see the following error and get no results (this works ok in Beam 2.24.0)
       
      Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.NullPointerException
              at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:36)
              at org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn$DoFnInvoker.invokeNewWatermarkEstimator(Unknown Source)
              at org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems$ProcessFn.processElement(SplittableParDoViaKeyedWorkItems.java:439)
              at org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems$ProcessFn$DoFnInvoker.invokeProcessElement(Unknown Source)
              at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:227)
              at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:183)
              at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62)
              at org.apache.beam.runners.flink.translation.wrappers.streaming.SplittableDoFnOperator.fireTimer(SplittableDoFnOperator.java:171)
              at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$FlinkTimerInternals.processPendingProcessingTimeTimers(DoFnOperator.java:1317)
              at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.close(DoFnOperator.java:575)
              at org.apache.beam.runners.flink.translation.wrappers.streaming.SplittableDoFnOperator.close(SplittableDoFnOperator.java:179)
              at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:618)
              at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$afterInvoke$1(StreamTask.java:498)
              at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
              at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:496)
              at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:477)
              at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
              at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
              at java.lang.Thread.run(Thread.java:748)
      Caused by: java.lang.NullPointerException
              at org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn.ensureTimestampWithinBounds(Read.java:541)
              at org.apache.beam.sdk.io.Read$UnboundedSourceAsSDFWrapperFn.newWatermarkEstimator(Read.java:552)

       

      Attachments

        Issue Links

          Activity

            People

              lcwik Luke Cwik
              iemejia Ismaël Mejía
              Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

                Created:
                Updated:
                Resolved: