[
https://issues.apache.org/jira/browse/FLINK-32626?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Austin Cawley-Edwards updated FLINK-32626:
------------------------------------------
Description:
The current `GET /jobs/:jobid/savepoints/:triggerid` API endpoint [1], when
given *either* a Job ID or a Trigger ID that does not exist it will always
respond with an exception that indicates {*}the Savepoint doesn't exist{*},
like:
{code:java}
{"errors":["org.apache.flink.runtime.rest.handler.RestHandlerException: There
is no savepoint operation with triggerId=TRIGGER ID for job JOB ID.\n\tat
org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.maybeCreateNotFoundError(SavepointHandlers.java:325)\n\tat
org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.lambda$handleRequest$1(SavepointHandlers.java:308)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:260)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1275)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:45)\n\tat
akka.dispatch.OnComplete.internal(Future.scala:299)\n\tat
akka.dispatch.OnComplete.internal(Future.scala:297)\n\tat
akka.dispatch.japi$CallbackBridge.apply(Future.scala:224)\n\tat
akka.dispatch.japi$CallbackBridge.apply(Future.scala:221)\n\tat
scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65)\n\tat
scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:72)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:288)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:288)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:288)\n\tat
akka.pattern.PromiseActorRef.$bang(AskSupport.scala:622)\n\tat
akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:25)\n\tat
akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23)\n\tat
scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536)\n\tat
scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)\n\tat
scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)\n\tat
scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63)\n\tat
akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100)\n\tat
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n\tat
scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85)\n\tat
akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100)\n\tat
akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)\n\tat
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)\n\tat
java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)\n\tat
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)\n\tat
java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)\n\tat
java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)\n\tat
java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)\n"]}
{code}
There are operational cases where it would be helpful to differentiate the Job
being not found vs. the Savepoint being not found.
For example, if a Job doesn't exist, we can stop trying to take a Savepoint on
it.
Additionally, we could consider removing the full stack trace as it doesn't add
value for these cases.
[1]:
[https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/rest_api/#jobs-jobid-savepoints-triggerid]
was:
The current `GET /jobs/:jobid/savepoints/:triggerid` API endpoint [1], when
given *either* a Job ID or a Trigger ID that does not exist, it will respond
with an exception that indicates the Savepoint doesn't exist, like:
{code:java}
{"errors":["org.apache.flink.runtime.rest.handler.RestHandlerException: There
is no savepoint operation with triggerId=TRIGGER ID for job JOB ID.\n\tat
org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.maybeCreateNotFoundError(SavepointHandlers.java:325)\n\tat
org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.lambda$handleRequest$1(SavepointHandlers.java:308)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:260)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1275)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)\n\tat
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)\n\tat
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:45)\n\tat
akka.dispatch.OnComplete.internal(Future.scala:299)\n\tat
akka.dispatch.OnComplete.internal(Future.scala:297)\n\tat
akka.dispatch.japi$CallbackBridge.apply(Future.scala:224)\n\tat
akka.dispatch.japi$CallbackBridge.apply(Future.scala:221)\n\tat
scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65)\n\tat
scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:72)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:288)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:288)\n\tat
scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:288)\n\tat
akka.pattern.PromiseActorRef.$bang(AskSupport.scala:622)\n\tat
akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:25)\n\tat
akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23)\n\tat
scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536)\n\tat
scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)\n\tat
scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)\n\tat
scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63)\n\tat
akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100)\n\tat
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n\tat
scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85)\n\tat
akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100)\n\tat
akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)\n\tat
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)\n\tat
java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)\n\tat
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)\n\tat
java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)\n\tat
java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)\n\tat
java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)\n"]}
{code}
There are operational cases where it would be helpful to differentiate the Job
being not found vs. the Savepoint being not found.
For example, if a Job doesn't exist, we can stop trying to take a Savepoint on
it.
Additionally, we could consider removing the full stack trace as it doesn't add
value for these cases.
[1]:
[https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/rest_api/#jobs-jobid-savepoints-triggerid]
> Distinguish non-existent job from non-existent savepoint in Get Savepoint
> REST API
> ----------------------------------------------------------------------------------
>
> Key: FLINK-32626
> URL: https://issues.apache.org/jira/browse/FLINK-32626
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / REST
> Affects Versions: 1.17.1
> Reporter: Austin Cawley-Edwards
> Priority: Not a Priority
>
> The current `GET /jobs/:jobid/savepoints/:triggerid` API endpoint [1], when
> given *either* a Job ID or a Trigger ID that does not exist it will always
> respond with an exception that indicates {*}the Savepoint doesn't exist{*},
> like:
> {code:java}
> {"errors":["org.apache.flink.runtime.rest.handler.RestHandlerException: There
> is no savepoint operation with triggerId=TRIGGER ID for job JOB ID.\n\tat
> org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.maybeCreateNotFoundError(SavepointHandlers.java:325)\n\tat
>
> org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers$SavepointStatusHandler.lambda$handleRequest$1(SavepointHandlers.java:308)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
>
> org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:260)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
>
> org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1275)\n\tat
>
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)\n\tat
>
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)\n\tat
>
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat
>
> java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)\n\tat
>
> org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:45)\n\tat
> akka.dispatch.OnComplete.internal(Future.scala:299)\n\tat
> akka.dispatch.OnComplete.internal(Future.scala:297)\n\tat
> akka.dispatch.japi$CallbackBridge.apply(Future.scala:224)\n\tat
> akka.dispatch.japi$CallbackBridge.apply(Future.scala:221)\n\tat
> scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
> org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65)\n\tat
>
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:72)\n\tat
>
> scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:288)\n\tat
>
> scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:288)\n\tat
>
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:288)\n\tat
> akka.pattern.PromiseActorRef.$bang(AskSupport.scala:622)\n\tat
> akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:25)\n\tat
>
> akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23)\n\tat
> scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536)\n\tat
> scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)\n\tat
> scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)\n\tat
> scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)\n\tat
> akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63)\n\tat
>
> akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100)\n\tat
> scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n\tat
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85)\n\tat
> akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100)\n\tat
> akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)\n\tat
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)\n\tat
>
> java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)\n\tat
>
> java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)\n\tat
>
> java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)\n\tat
>
> java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)\n\tat
>
> java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)\n"]}
> {code}
> There are operational cases where it would be helpful to differentiate the
> Job being not found vs. the Savepoint being not found.
> For example, if a Job doesn't exist, we can stop trying to take a Savepoint
> on it.
>
> Additionally, we could consider removing the full stack trace as it doesn't
> add value for these cases.
>
> [1]:
> [https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/rest_api/#jobs-jobid-savepoints-triggerid]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
