Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-20418

NPE in IteratorSourceReader

    XMLWordPrintableJSON

    Details

      Description

      With the following job

      	@Test
      	public void testNpe() throws Exception {
      		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      		env.setParallelism(1);
      		env.setRestartStrategy(new NoRestartStrategyConfiguration());
      		env.enableCheckpointing(50, CheckpointingMode.EXACTLY_ONCE);
      		env
      			.fromSequence(0, 100)
      			.map(x -> {
      				Thread.sleep(10);
      				return x;
      			})
      			.addSink(new DiscardingSink<>());
      		env.execute();
      	}
      

      I (always) get an exception like this:

      ...
      Caused by: java.lang.Exception: Could not perform checkpoint 1 for operator Source: Sequence Source -> Map -> Sink: Unnamed (1/1)#0.
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:866)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$8(StreamTask.java:831)
      	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
      	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:78)
      	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:283)
      	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:184)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:575)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:539)
      	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:722)
      	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:547)
      	at java.lang.Thread.run(Thread.java:748)
      Caused by: org.apache.flink.runtime.checkpoint.CheckpointException: Could not complete snapshot 1 for operator Source: Sequence Source -> Map -> Sink: Unnamed (1/1)#0. Failure reason: Checkpoint was declined.
      	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:226)
      	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:158)
      	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:343)
      	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:603)
      	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:529)
      	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:496)
      	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:266)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$9(StreamTask.java:924)
      	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:914)
      	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:857)
      	... 10 more
      Caused by: java.lang.NullPointerException
      	at org.apache.flink.api.connector.source.lib.util.IteratorSourceReader.snapshotState(IteratorSourceReader.java:132)
      	at org.apache.flink.streaming.api.operators.SourceOperator.snapshotState(SourceOperator.java:264)
      	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:197)
      	... 20 more
      

      Adding a null check solves the issue. But if I then change sleep time from 10 to 50 I get

      Caused by: java.lang.IllegalArgumentException: 'from' must be <= 'to'
      	at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:142)
      	at org.apache.flink.api.connector.source.lib.NumberSequenceSource$NumberSequenceSplit.<init>(NumberSequenceSource.java:142)
      	at org.apache.flink.api.connector.source.lib.NumberSequenceSource$NumberSequenceSplit.getUpdatedSplitForIterator(NumberSequenceSource.java:169)
      	at org.apache.flink.api.connector.source.lib.NumberSequenceSource$NumberSequenceSplit.getUpdatedSplitForIterator(NumberSequenceSource.java:135)
      	at org.apache.flink.api.connector.source.lib.util.IteratorSourceReader.snapshotState(IteratorSourceReader.java:135)
      	at org.apache.flink.streaming.api.operators.SourceOperator.snapshotState(SourceOperator.java:264)
      	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:197)
      	... 20 more
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                AHeise Arvid Heise
                Reporter:
                roman_khachatryan Roman Khachatryan
              • Votes:
                0 Vote for this issue
                Watchers:
                3 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: