You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Velu Mitwa <ve...@gmail.com> on 2018/01/03 15:02:55 UTC
Fwd: Queryable State Client - Actor Not found Exception
Hi,
I am running a Flink Job which uses the Queryable State feature of Apache
Flink(1.3.2). I was able to do that in local mode. When I try to do that in
Cluster mode (Yarn Session), I am getting Actor not found Exception.
Please help me to understand what is missing.
*Exception Trace*
Query failed because of the following Exception:
akka.actor.ActorNotFound: Actor not found for: ActorSelection[Anchor(akka.
tcp://flink@my-machine:52650/), Path(/user/jobmanager)]
at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(
ActorSelection.scala:65)
at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(
ActorSelection.scala:63)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(
BatchingExecutor.scala:55)
at akka.dispatch.BatchingExecutor$Batch.run(
BatchingExecutor.scala:73)
at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.
unbatchedExecute(Future.scala:74)
at akka.dispatch.BatchingExecutor$class.execute(BatchingExecutor.
scala:120)
at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.
execute(Future.scala:73)
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.
scala:40)
at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(
Promise.scala:248)
at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(
AskSupport.scala:334)
at akka.actor.Scheduler$$anon$7.run(Scheduler.scala:117)
at scala.concurrent.Future$InternalCallbackExecutor$.
scala$concurrent$Future$InternalCallbackExecutor$$
unbatchedExecute(Future.scala:694)
at scala.concurrent.Future$InternalCallbackExecutor$.
execute(Future.scala:691)
at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(
Scheduler.scala:474)
at akka.actor.LightArrayRevolverScheduler$$anon$8.executeBucket$1(
Scheduler.scala:425)
at akka.actor.LightArrayRevolverScheduler$$
anon$8.nextTick(Scheduler.scala:429)
at akka.actor.LightArrayRevolverScheduler$$
anon$8.run(Scheduler.scala:381)
at java.lang.Thread.run(Thread.java:745)
*Client Creation Snippet *
* Configuration config = new Configuration();*
* config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY,
jobManagerHost);*
* config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY,
jobManagerPort);*
* final HighAvailabilityServices highAvailabilityServices =
HighAvailabilityServicesUtils*
* .createHighAvailabilityServices(config,
Executors.newSingleThreadScheduledExecutor(),*
*
HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);*
* this.client = new QueryableStateClient(config,
highAvailabilityServices);*
* }*
Re: Queryable State Client - Actor Not found Exception
Posted by Velu Mitwa <ve...@gmail.com>.
Thank you Aljoscha.
I am able to Query state when I use the hostname of Job Manager instead of
its IP Address. But I couldn't understand why it is not working if I give
IP address.
On Thu, Jan 4, 2018 at 6:42 PM, Aljoscha Krettek <al...@apache.org>
wrote:
> Hi,
>
> Is my-machine:52650 the correct address for the JobManager running in
> YARN? Also, Queryable State in Flink 1.3.2 is not easy to get to work when
> you use YARN with HA mode.
>
> Best,
> Aljoscha
>
> On 3. Jan 2018, at 16:02, Velu Mitwa <ve...@gmail.com> wrote:
>
> Hi,
> I am running a Flink Job which uses the Queryable State feature of Apache
> Flink(1.3.2). I was able to do that in local mode. When I try to do that in
> Cluster mode (Yarn Session), I am getting Actor not found Exception.
>
> Please help me to understand what is missing.
>
> *Exception Trace*
>
>
> Query failed because of the following Exception:
> akka.actor.ActorNotFound: Actor not found for: ActorSelection[Anchor(akka.
> tcp://flink@my-machine:52650/), Path(/user/jobmanager)]
> at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(ActorS
> election.scala:65)
> at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(ActorS
> election.scala:63)
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
> at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(Ba
> tchingExecutor.scala:55)
> at akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.
> scala:73)
> at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.
> unbatchedExecute(Future.scala:74)
> at akka.dispatch.BatchingExecutor$class.execute(
> BatchingExecutor.scala:120)
> at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.
> execute(Future.scala:73)
> at scala.concurrent.impl.CallbackRunnable.executeWithValue(
> Promise.scala:40)
> at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Pro
> mise.scala:248)
> at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupp
> ort.scala:334)
> at akka.actor.Scheduler$$anon$7.run(Scheduler.scala:117)
> at scala.concurrent.Future$InternalCallbackExecutor$.scala$
> concurrent$Future$InternalCallbackExecutor$$unbatchedExecute
> (Future.scala:694)
> at scala.concurrent.Future$InternalCallbackExecutor$.execute(
> Future.scala:691)
> at akka.actor.LightArrayRevolverScheduler$TaskHolder.
> executeTask(Scheduler.scala:474)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.
> executeBucket$1(Scheduler.scala:425)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.nextTick(
> Scheduler.scala:429)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.run(
> Scheduler.scala:381)
> at java.lang.Thread.run(Thread.java:745)
>
> *Client Creation Snippet *
>
> * Configuration config = new Configuration();*
> * config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY,
> jobManagerHost);*
> * config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY,
> jobManagerPort);*
>
> * final HighAvailabilityServices highAvailabilityServices =
> HighAvailabilityServicesUtils*
> * .createHighAvailabilityServices(config,
> Executors.newSingleThreadScheduledExecutor(),*
> *
> HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);*
>
> * this.client = new QueryableStateClient(config,
> highAvailabilityServices);*
> * }*
>
>
>
>
Re: Queryable State Client - Actor Not found Exception
Posted by Aljoscha Krettek <al...@apache.org>.
Hi,
Is my-machine:52650 the correct address for the JobManager running in YARN? Also, Queryable State in Flink 1.3.2 is not easy to get to work when you use YARN with HA mode.
Best,
Aljoscha
> On 3. Jan 2018, at 16:02, Velu Mitwa <ve...@gmail.com> wrote:
>
> Hi,
> I am running a Flink Job which uses the Queryable State feature of Apache Flink(1.3.2). I was able to do that in local mode. When I try to do that in Cluster mode (Yarn Session), I am getting Actor not found Exception.
>
> Please help me to understand what is missing.
>
> Exception Trace
>
> Query failed because of the following Exception:
> akka.actor.ActorNotFound: Actor not found for: ActorSelection[Anchor(akka.tcp://flink@my-machine:52650/), Path(/user/jobmanager)]
> at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(ActorSelection.scala:65)
> at akka.actor.ActorSelection$$anonfun$resolveOne$1.apply(ActorSelection.scala:63)
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
> at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
> at akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:73)
> at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.unbatchedExecute(Future.scala:74)
> at akka.dispatch.BatchingExecutor$class.execute(BatchingExecutor.scala:120)
> at akka.dispatch.ExecutionContexts$sameThreadExecutionContext$.execute(Future.scala:73)
> at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
> at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
> at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:334)
> at akka.actor.Scheduler$$anon$7.run(Scheduler.scala:117)
> at scala.concurrent.Future$InternalCallbackExecutor$.scala$concurrent$Future$InternalCallbackExecutor$$unbatchedExecute(Future.scala:694)
> at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:691)
> at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(Scheduler.scala:474)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.executeBucket$1(Scheduler.scala:425)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.nextTick(Scheduler.scala:429)
> at akka.actor.LightArrayRevolverScheduler$$anon$8.run(Scheduler.scala:381)
> at java.lang.Thread.run(Thread.java:745)
>
> Client Creation Snippet
>
> Configuration config = new Configuration();
> config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, jobManagerHost);
> config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, jobManagerPort);
>
> final HighAvailabilityServices highAvailabilityServices = HighAvailabilityServicesUtils
> .createHighAvailabilityServices(config, Executors.newSingleThreadScheduledExecutor(),
> HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);
>
> this.client = new QueryableStateClient(config, highAvailabilityServices);
> }
>