I m having a very weird issue with stop with savepoint on Fl Apache Flink #troubleshooting

I’m having a very weird issue with stop with savep...

Bogdan Chifor

10/09/2023, 7:16 PM

I’m having a very weird issue with stop with savepoint on Flink 1.15.4. In short, when I try to stop with savepoint via the REST api, my request immediately fails with a rather generic error: `org.apache.flink.runtime.checkpoint.CheckpointException: Task has failed`This usually happens within 5-6s. None of the operators in the job acknowledges the checkpoint. I should mention that I can trigger a savepoint via REST using the

savepoints

endpoint, I can even use the

cancel: true

payload field to actually stop the job with a savepoint, and that works just fine. It’s only when hitting the

jobs/{job_id}/stop

endpoint that I run into this issue. I’ll provide more details in the 🧵 , but I’m curious if anyone else ran into this issue before.

Bogdan Chifor

10/09/2023, 7:17 PM

URL:

POST http://<JM_URL>/jobs/00000000000000000000000000000000/stop

Payload:

{"drain":false,"targetDirectory":"s3p://...","formatType":"CANONICAL"}

Bogdan Chifor

10/09/2023, 7:18 PM

I’ve also tried adding a triggerId to the payload, with the same result

Bogdan Chifor

10/09/2023, 7:18 PM

Copy code

org.apache.flink.runtime.checkpoint.CheckpointException: Task has failed.
	at org.apache.flink.runtime.messages.checkpoint.SerializedCheckpointException.unwrap(SerializedCheckpointException.java:51)
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveDeclineMessage(CheckpointCoordinator.java:1013)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$declineCheckpoint$2(ExecutionGraphHandler.java:103)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$processCheckpointCoordinatorMessage$3(ExecutionGraphHandler.java:119)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.runtime.checkpoint.CheckpointException: Task name with subtask : Source: track-stream -> (kafka-bird-tracks-watermark, telemetry-parser) (5/18)#0 Failure reason: Task has failed.
	at org.apache.flink.runtime.taskmanager.Task.declineCheckpoint(Task.java:1388)
	at org.apache.flink.runtime.taskmanager.Task.lambda$triggerCheckpointBarrier$3(Task.java:1331)
	at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:343)
Caused by: java.util.concurrent.CompletionException: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(Unknown Source)
	... 3 more
Caused by: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException
	at org.apache.flink.streaming.connectors.kafka.internals.Handover.close(Handover.java:177)
	at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.cancel(KafkaFetcher.java:164)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:945)
	at org.apache.flink.streaming.api.operators.StreamSource.stop(StreamSource.java:128)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.stopOperatorForStopWithSavepoint(SourceStreamTask.java:305)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.lambda$triggerStopWithSavepointAsync$1(SourceStreamTask.java:285)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:338)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:324)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:807)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:756)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
	at java.base/java.lang.Thread.run(Unknown Source)

Bogdan Chifor

10/09/2023, 7:18 PM

Full error stack

Martijn Visser

10/09/2023, 7:43 PM

Are you able to upgrade to a newer Flink version?

Bogdan Chifor

10/09/2023, 7:46 PM

I’m seeing this error because we’re upgrading from 1.14.6 to 1.15.4. This upgrade was not very straightforward, a bunch of libraries changed and also some semantics around savepoints and checkpoints have also changed. I haven’t yet looked into bumping to version 16, I guess I could try to see if that solves my issue, but I’d be first curious to try and understand what’s currently going on

Martijn Visser

10/09/2023, 8:09 PM

Did the release notes help before doing the upgrade?

Danny Cranmer

10/09/2023, 8:11 PM

I see you are using FlinkKafkaConsumer. There are a few bugs that prevent graceful close, including https://issues.apache.org/jira/browse/FLINK-28758. The recommendation is to upgrade to the new KafkaSource.

➕ 1

👀 1

Bogdan Chifor

10/10/2023, 12:06 AM

@Martijn Visser release notes were great!

Bogdan Chifor

10/10/2023, 12:06 AM

Thank you @Danny Cranmer will try that.

Bogdan Chifor

10/10/2023, 12:39 AM

Just out of curiosity, would you know if there is a semantic difference between

stop with savepoint

and

trigger savepoint and cancel

Bogdan Chifor

10/10/2023, 5:37 AM

Based on the my interpretation of the REST API documentation, the second

trigger savepoint and cancel

(POST on /jobs/:jobid/savepoints
with
cancel-job: true
) will trigger the savepoint, and then cancel the job, in a non atomic operation.

Martijn Visser

10/10/2023, 7:06 AM

Another action for stopping a job is stop. It is a more graceful way of stopping a running streaming job as the stop flows from source to sink. When the user requests to stop a job, all sources will be requested to send the last checkpoint barrier that will trigger a savepoint, and after the successful completion of that savepoint, they will finish by calling their cancel() method.

Another important bit:

The
--withSavepoint
flag allows creating a savepoint as part of the job cancellation. This feature is deprecated. Use the stop action instead.

3 Views

Open in Slack

Previous Next