Hi team, when I upgrade Flink jobs to 1.16, many of the jobs fail to perform further deployment/upgrade due to error

Caused by: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException

. I have tried to increase kafka request time out for the consumer, but it does not have any effect. and the issue is still there. The flink jobs are running under standalone mode and are managed by apache flink operator 1.3.1. During upgrade, the stop and savepoint operation hits the following but manually trigger savepoints are totally OK. Previously there was discussion mentioned it can be resolved by changing the kafka consumer/producer to kafka source and sink. However we have > 50 flink jobs so need time to change all of them. Anybody know if there is workaround for this issue? Thanks!

2023-02-22 22:25:42
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:1036)
	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: org.apache.flink.runtime.checkpoint.CheckpointException: Task name with subtask : Source: Event Source -> Extract Value From Kafka Message -> Add Datamart Metadata -> Filter Datamart Orgs And Stores (1/1)#2 Failure reason: Task has failed.
	at org.apache.flink.runtime.taskmanager.Task.declineCheckpoint(Task.java:1375)
	at org.apache.flink.runtime.taskmanager.Task.lambda$triggerCheckpointBarrier$3(Task.java:1318)
	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:344)
Caused by: java.util.concurrent.CompletionException: 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: 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:306)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.lambda$triggerStopWithSavepointAsync$1(SourceStreamTask.java:286)
	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.runMail(MailboxProcessor.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Unknown Source)

I think in this case, you can still use flink cli to do your deployment, if you can keep using flink cli for your CI/CD for the moment and wait the kafka source/sink udpate. It could work

Thanks @Yang LI for the suggestion! We only use CI/CD for the deployment so may have to change to use source and sink.