Get Flink ExecutionGraph Programmatically

[Chawla,Sumit]

Hi All

I am trying to get JOB  accumulators.  ( I am aware that I can get the accumulators through REST APIs as well, but i wanted to avoid JSON parsing).  

Looking at JobAccumulatorsHandler i am trying to get execution graph for currently running job.  Following is my code:

  InetSocketAddress initialJobManagerAddress=new InetSocketAddress(hostName,port);
            InetAddress ownHostname;
            ownHostname= ConnectionUtils.findConnectingAddress(initialJobManagerAddress,2000,400);

            ActorSystem actorSystem= AkkaUtils.createActorSystem(configuration,
                    new Some(new Tuple2<String,Object>(ownHostname.getCanonicalHostName(),0)));

            FiniteDuration timeout= FiniteDuration.apply(10, TimeUnit.SECONDS);

            ActorGateway akkaActorGateway= LeaderRetrievalUtils.retrieveLeaderGateway(
                    LeaderRetrievalUtils.createLeaderRetrievalService(configuration),
                    actorSystem,timeout
            );


            Future<Object> future=akkaActorGateway.ask(new RequestJobDetails(true,false),timeout);

            MultipleJobsDetails result=(MultipleJobsDetails) Await.result(future,timeout);
            ExecutionGraphHolder executionGraphHolder=new ExecutionGraphHolder(timeout);
            LOG.info(result.toString());
            for(JobDetails detail:result.getRunningJobs()){
                LOG.info(detail.getJobName() + "  ID " + detail.getJobId());
                ExecutionGraph executionGraph=executionGraphHolder.getExecutionGraph(detail.getJobId(),akkaActorGateway);
                LOG.info("Accumulators " + executionGraph.aggregateUserAccumulators());
            }

 

However, i am receiving following error in Flink:

2016-09-20T14:19:53,201 [flink-akka.actor.default-dispatcher-3] nobody ERROR akka.remote.EndpointWriter - Transient association error (association remains live)

java.io.NotSerializableException: org.apache.flink.runtime.checkpoint.CheckpointCoordinator

        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_92]

        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) ~[?:1.8.0_92]

        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply$mcV$sp(Serializer.scala:129) ~[akka-actor_2.10-2.3.7.jar:?]

        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129) ~[akka-actor_2.10-2.3.7.jar:?]

        at akka.serialization.JavaSerializer$$anonfun$toBinary$1.apply(Serializer.scala:129) ~[akka-actor_2.10-2.3.7.jar:?]

        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57) ~[scala-library-2.10.5.jar:?]

        at akka.serialization.JavaSerializer.toBinary(Serializer.scala:129) ~[akka-actor_2.10-2.3.7.jar:?]

        at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36) ~[akka-remote_2.10-2.3.7.jar:?]

        at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845) ~[akka-remote_2.10-2.3.7.jar:?]

        at akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:845) ~[akka-remote_2.10-2.3.7.jar:?]

        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57) ~[scala-library-2.10.5.jar:?]

        at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:844) ~[akka-remote_2.10-2.3.7.jar:?]

Any reason why its failing? This code works when invoked through WebRuntimeMonitor.