OSDir


[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Failure restarting Flink 1.5.0 job from checkpoint


Hi Aljoscha,

Yes, looks that way, thanks the issue reference - I’d checked Jira few days ago, looks like FLINK-9458 was added very recently :)

I’ll follow up in Jira to see if a small code snippet would be useful.

— Ken

> On May 31, 2018, at 1:17 AM, Aljoscha Krettek <aljoscha@xxxxxxxxxx> wrote:
> 
> Hi Ken,
> 
> I think you might have independently discovered this issue: https://issues.apache.org/jira/browse/FLINK-9458 <https://issues.apache.org/jira/browse/FLINK-9458>
> 
> Best,
> Aljoscha
> 
>> On 31. May 2018, at 01:46, Ken Krugler <kkrugler_lists@xxxxxxxxxxxx> wrote:
>> 
>> Hi devs,
>> 
>> I coded up a simple iteration that uses a KeyedProcessFunction, as a way of showing how to use timers to do state iteration.
>> 
>> This worked fine, but then I wanted to try out checkpoints. I modified the KeyedProcessFunction to throw an exception after a fixed number of calls.
>> 
>> When this happens, it puts my job into a loop, where restarting the job fails with a NullPointerException:
>> 
>> 18/05/30 16:38:40 DEBUG executiongraph.ExecutionGraph:1496 - Try to restart or fail the job Flink Streaming Job (f144fd0fb301db0ae14c7b991a25b353) if no longer possible.
>> java.lang.RuntimeException: Example of a failure triggering a job restart
>> 	at com.scaleunlimited.flinksnippets.examples.IterationWithProcessFunctionTimers$MyKeyedProcessFunction.processElement(IterationWithProcessFunctionTimers.java:74)
>> 	at com.scaleunlimited.flinksnippets.examples.IterationWithProcessFunctionTimers$MyKeyedProcessFunction.processElement(IterationWithProcessFunctionTimers.java:1)
>> 	at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:85)
>> 	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)
>> 	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:103)
>> 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
>> 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)
>> 	at java.lang.Thread.run(Thread.java:748)
>> 18/05/30 16:38:40 INFO executiongraph.ExecutionGraph:1375 - Job Flink Streaming Job (f144fd0fb301db0ae14c7b991a25b353) switched from state FAILING to RESTARTING.
>> 18/05/30 16:38:40 INFO executiongraph.ExecutionGraph:1506 - Restarting the job Flink Streaming Job (f144fd0fb301db0ae14c7b991a25b353).
>> 18/05/30 16:38:40 WARN executiongraph.ExecutionGraph:1273 - Failed to restart the job.
>> java.lang.NullPointerException
>> 	at org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint.isAssignedAndAlive(CoLocationConstraint.java:104)
>> 	at org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup.resetConstraints(CoLocationGroup.java:119)
>> 	at org.apache.flink.runtime.executiongraph.ExecutionGraph.restart(ExecutionGraph.java:1247)
>> 	at org.apache.flink.runtime.executiongraph.restart.ExecutionGraphRestartCallback.triggerFullRecovery(ExecutionGraphRestartCallback.java:59)
>> 	at org.apache.flink.runtime.executiongraph.restart.FixedDelayRestartStrategy$1.run(FixedDelayRestartStrategy.java:68)
>> 	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:748)
>> 
>> CoLocationContraint.java:104 is this one line function:
>> 
>> 	public boolean isAssignedAndAlive() {
>> 		return lockedLocation != null && sharedSlot.isAlive();
>> 	}
>> 
>> So I have to assume sharedSlot is null - I don’t know if that’s valid, or if this means that the constraint is being used before setSharedSlot() is called.
>> 
>> In any case, this same chunk of logging output repeats immediately, ad infinitum.
>> 
>> Is there something else I should try to track down what’s going on?
>> 
>> Thanks,
>> 
>> — Ken
>> 
>> PS - checkpointing is set up via:
>> 
>>       final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(2);
>>       env.setParallelism(2);
>>       env.enableCheckpointing(100L, CheckpointingMode.AT_LEAST_ONCE, true);
>>       env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);
>> 
>> And the skeleton of the simple workflow is:
>> 
>>       IterativeStream<String> iter = env.addSource(source).iterate(1000L);
>>       DataStream<String> updated = iter.keyBy(new MyKeySelector()).process(new MyKeyedProcessFunction());
>>       iter.closeWith(updated).print();

--------------------------
Ken Krugler
+1 530-210-6378
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra