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

Race condition between KafkaIO.UnboundedKafkaReader.getWatermark() and KafkaIO.UnboundedKafkaReader.advance()

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: P2
    • Resolution: Fixed
    • Affects Version/s: 2.0.0, 2.1.0
    • Fix Version/s: 2.2.0
    • Component/s: sdk-java-core
    • Labels:
      None

      Description

      getWatermark() looks like this:

          @Override
          public Instant getWatermark() {
            if (curRecord == null) {
              LOG.debug("{}: getWatermark() : no records have been read yet.", name);
              return initialWatermark;
            }
      
            return source.spec.getWatermarkFn() != null
                ? source.spec.getWatermarkFn().apply(curRecord) : curTimestamp;
          }
      

      advance() has code in it that looks like this:

                curRecord = null; // user coders below might throw.
      
                // apply user deserializers.
                // TODO: write records that can't be deserialized to a "dead-letter" additional output.
                KafkaRecord<K, V> record = new KafkaRecord<K, V>(
                    rawRecord.topic(),
                    rawRecord.partition(),
                    rawRecord.offset(),
                    consumerSpEL.getRecordTimestamp(rawRecord),
                    keyDeserializerInstance.deserialize(rawRecord.topic(), rawRecord.key()),
                    valueDeserializerInstance.deserialize(rawRecord.topic(), rawRecord.value()));
      
                curTimestamp = (source.spec.getTimestampFn() == null)
                    ? Instant.now() : source.spec.getTimestampFn().apply(record);
                curRecord = record;
      

      There's a race condition between these two blocks of code which is exposed at the very least in the FlinkRunner, which calls getWatermark() periodically from a timer.

      The symptom of the race condition is a stack trace that looks like this (SDK 2.0.0):

      Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
      	at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:910)
      	at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:853)
      	at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:853)
      	at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
      	at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
      	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
      	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397)
      	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
      Caused by: TimerException{java.lang.NullPointerException}
      	at org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:220)
      	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
      	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
      	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      	at java.lang.Thread.run(Thread.java:745)
      Caused by: java.lang.NullPointerException
      	at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:568)
      	at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:565)
      	at org.apache.beam.sdk.io.kafka.KafkaIO$UnboundedKafkaReader.getWatermark(KafkaIO.java:1210)
      	at org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.onProcessingTime(UnboundedSourceWrapper.java:431)
      	at org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:218)
      	... 7 more
      

      Based on inspecting the code, what is probably happening is that while advance() is executing:

      • Flink runner calls getWatermark()
      • getWatermark evaluates (curRecord == null) and it is false
      • advance() proceeds to set curRecord = null
      • The flink runner thread calls getWatermarkFn().apply(curRecord) which passes a null record into the custom watermark function
      • If that watermark function had been set with withWatermarkFn() (as suggested in the javadoc), then it's using the closure created in unwrapKafkaAndThen()
      • That calls record.getKV() and we get the NullPointerException

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                rangadi Raghu Angadi
                Reporter:
                wtanaka Wesley Tanaka
              • Votes:
                0 Vote for this issue
                Watchers:
                5 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: