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

Querying job overview in the REST API fails

    XMLWordPrintableJSON

Details

    • Bug
    • Status: Closed
    • Major
    • Resolution: Duplicate
    • 1.14.3
    • None
    • Runtime / Network
    • None

    Description

      Hello,

      In my setup there are three servers in a standalone the Flink cluster using ZooKeeper HA Services running Flink 1.14.3. There's one TaskManager and one JobManager on all servers. ZooKeeper is running on all servers. All the servers have been just started and one simple job has been deployed.

      One of the JobManagers is the leading JobManager.

       

      If I query (curl) the leading job manager with /v1/jobs/overview, the response is correct.

      But if I query a JobManager that is not the leading JobManager, the HTTP request fails and the following can be seen in the logs.

      2022-03-26 10:12:55,539 ERROR org.apache.flink.runtime.rest.handler.job.JobsOverviewHandler [] - Unhandled exception.
      org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException: Failed to serialize the result for RPC call : requestMultipleJobDetails.
              at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.serializeRemoteResultAndVerifySize(AkkaRpcActor.java:417) ~[?:?]
              at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$sendAsyncResponse$2(AkkaRpcActor.java:373) ~[?:?]
              at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:836) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:811) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) ~[?:1.8.0_261]
              at org.apache.flink.util.concurrent.FutureUtils$ResultConjunctFuture.handleCompletedFuture(FutureUtils.java:858) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at org.apache.flink.util.concurrent.FutureUtils$ResultConjunctFuture.lambda$new$0(FutureUtils.java:876) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) ~[?:1.8.0_261]
              at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:258) ~[?:?]
              at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) ~[?:1.8.0_261]
              at org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1389) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93) ~[?:?]
              at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[?:?]
              at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92) ~[?:?]
              at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_261]
              at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) ~[?:1.8.0_261]
              at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:47) ~[?:?]
              at akka.dispatch.OnComplete.internal(Future.scala:300) ~[?:?]
              at akka.dispatch.OnComplete.internal(Future.scala:297) ~[?:?]
              at akka.dispatch.japi$CallbackBridge.apply(Future.scala:224) ~[?:?]
              at akka.dispatch.japi$CallbackBridge.apply(Future.scala:221) ~[?:?]
              at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65) ~[?:?]
              at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:68) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:284) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:284) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:621) ~[?:?]
              at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:24) ~[?:?]
              at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23) ~[?:?]
              at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:532) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:29) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:29) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63) ~[?:?]
              at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100) ~[?:?]
              at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12) ~[?:?]
              at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100) ~[?:?]
              at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49) ~[?:?]
              at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48) ~[?:?]
              at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [?:1.8.0_261]
              at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) [?:1.8.0_261]
              at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) [?:1.8.0_261]
              at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172) [?:1.8.0_261]
      Caused by: java.io.NotSerializableException: java.util.HashMap$Values
              at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184) ~[?:1.8.0_261]
              at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) ~[?:1.8.0_261]
              at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) ~[?:1.8.0_261]
              at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) ~[?:1.8.0_261]
              at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_261]
              at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) ~[?:1.8.0_261]
              at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:632) ~[flink-dist_2.11-1.14.3.jar:1.14.3]
              at org.apache.flink.runtime.rpc.akka.AkkaRpcSerializedValue.valueOf(AkkaRpcSerializedValue.java:66) ~[?:?]
              at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.serializeRemoteResultAndVerifySize(AkkaRpcActor.java:400) ~[?:?]
              ... 53 more 

      Attachments

        Issue Links

          Activity

            People

              Unassigned Unassigned
              Peihui Peihui He
              Votes:
              0 Vote for this issue
              Watchers:
              1 Start watching this issue

              Dates

                Created:
                Updated:
                Resolved: