import akka.actor.Identify; //导入依赖的package包/类
@SuppressWarnings("checkstyle:IllegalCatch")
private void verifyActorReady(ActorRef actorRef) {
    // Sometimes we see messages go to dead letters soon after creation - it seems the actor isn't quite
    // in a state yet to receive messages or isn't actually created yet. This seems to happen with
    // actorSelection so, to alleviate it, we use an actorSelection and send an Identify message with
    // retries to ensure it's ready.

    Timeout timeout = new Timeout(100, TimeUnit.MILLISECONDS);
    Throwable lastError = null;
    Stopwatch sw = Stopwatch.createStarted();
    while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
        try {
            ActorSelection actorSelection = system.actorSelection(actorRef.path().toString());
            Future<Object> future = Patterns.ask(actorSelection, new Identify(""), timeout);
            ActorIdentity reply = (ActorIdentity)Await.result(future, timeout.duration());
            Assert.assertNotNull("Identify returned null", reply.getRef());
            return;
        } catch (Exception | AssertionError e) {
            Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
            lastError = e;
        }
    }

    throw new RuntimeException(lastError);
}
 

import akka.actor.Identify; //导入依赖的package包/类
@Test
public void testIdentifying() {
    getSystem().actorOf(ConstantEcho.props("foo"), "foo");

    final ActorSelection selection = getSystem().actorSelection("/user/foo");
    selection.tell(new Identify("identifyFoo"), testActor());

    final Object[] seq = receiveN(1);

    ActorIdentity identity = (ActorIdentity) seq[0];
    assertEquals("identifyFoo", identity.correlationId());

    identity.getRef().tell("baz", testActor());
    expectMsgEquals("foo");
}
 

import akka.actor.Identify; //导入依赖的package包/类
@Override
public void onReceive(Object msg) throws Exception {	
	if (msg instanceof ActorIdentity && isTarget((ActorIdentity) msg)) {
		this.owner.tell(msg, getSender());
		this.found = true;
	} if (msg.equals(START)) {
		this.found = false;
		getSelf().tell(LOOKUP, getSelf());
	} else if (!found && msg.equals(LOOKUP)) {
		ActorSelection selection =  this.getContext().actorSelection(path);
		selection.tell(new Identify(path), this.getSelf());
		getContext().system().scheduler().scheduleOnce(
				Duration.create(RETRY_PERIOD, TimeUnit.SECONDS),
				getSelf(), 
				LOOKUP, 
				getContext().dispatcher(), 
				getSelf());
	}
}
 

import akka.actor.Identify; //导入依赖的package包/类
private <C extends RpcGateway> CompletableFuture<C> connectInternal(
		final String address,
		final Class<C> clazz,
		Function<ActorRef, InvocationHandler> invocationHandlerFactory) {
	checkState(!stopped, "RpcService is stopped");

	LOG.debug("Try to connect to remote RPC endpoint with address {}. Returning a {} gateway.",
		address, clazz.getName());

	final ActorSelection actorSel = actorSystem.actorSelection(address);

	final Future<Object> identify = Patterns.ask(actorSel, new Identify(42), timeout.toMilliseconds());
	final Future<C> resultFuture = identify.map(new Mapper<Object, C>(){
		@Override
		public C checkedApply(Object obj) throws Exception {

			ActorIdentity actorIdentity = (ActorIdentity) obj;

			if (actorIdentity.getRef() == null) {
				throw new RpcConnectionException("Could not connect to rpc endpoint under address " + address + '.');
			} else {
				ActorRef actorRef = actorIdentity.getRef();

				InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef);

				// Rather than using the System ClassLoader directly, we derive the ClassLoader
				// from this class . That works better in cases where Flink runs embedded and all Flink
				// code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
				ClassLoader classLoader = AkkaRpcService.this.getClass().getClassLoader();

				@SuppressWarnings("unchecked")
				C proxy = (C) Proxy.newProxyInstance(
					classLoader,
					new Class<?>[]{clazz},
					invocationHandler);

				return proxy;
			}
		}
	}, actorSystem.dispatcher());

	return FutureUtils.toJava(resultFuture);
}
 

import akka.actor.Identify; //导入依赖的package包/类
/**
 * Tests that a failing job recovery won't cause other job recoveries to fail.
 */
@Test
public void testFailingJobRecovery() throws Exception {
	final FiniteDuration timeout = new FiniteDuration(10, TimeUnit.SECONDS);
	final FiniteDuration jobRecoveryTimeout = new FiniteDuration(0, TimeUnit.SECONDS);
	Deadline deadline = new FiniteDuration(1, TimeUnit.MINUTES).fromNow();
	final Configuration flinkConfiguration = new Configuration();
	UUID leaderSessionID = UUID.randomUUID();
	ActorRef jobManager = null;
	JobID jobId1 = new JobID();
	JobID jobId2 = new JobID();

	// set HA mode to zookeeper so that we try to recover jobs
	flinkConfiguration.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");

	try {
		final SubmittedJobGraphStore submittedJobGraphStore = mock(SubmittedJobGraphStore.class);

		SubmittedJobGraph submittedJobGraph = mock(SubmittedJobGraph.class);
		when(submittedJobGraph.getJobId()).thenReturn(jobId2);

		when(submittedJobGraphStore.getJobIds()).thenReturn(Arrays.asList(jobId1, jobId2));

		// fail the first job recovery
		when(submittedJobGraphStore.recoverJobGraph(eq(jobId1))).thenThrow(new Exception("Test exception"));
		// succeed the second job recovery
		when(submittedJobGraphStore.recoverJobGraph(eq(jobId2))).thenReturn(submittedJobGraph);

		final TestingLeaderElectionService myLeaderElectionService = new TestingLeaderElectionService();

		final Collection<JobID> recoveredJobs = new ArrayList<>(2);

		BlobServer blobServer = mock(BlobServer.class);
		Props jobManagerProps = Props.create(
			TestingFailingHAJobManager.class,
			flinkConfiguration,
			TestingUtils.defaultExecutor(),
			TestingUtils.defaultExecutor(),
			mock(InstanceManager.class),
			mock(Scheduler.class),
			blobServer,
			new BlobLibraryCacheManager(blobServer, FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST, new String[0]),
			ActorRef.noSender(),
			new FixedDelayRestartStrategy.FixedDelayRestartStrategyFactory(Int.MaxValue(), 100),
			timeout,
			myLeaderElectionService,
			submittedJobGraphStore,
			mock(CheckpointRecoveryFactory.class),
			jobRecoveryTimeout,
			UnregisteredMetricGroups.createUnregisteredJobManagerMetricGroup(),
			recoveredJobs).withDispatcher(CallingThreadDispatcher.Id());

		jobManager = system.actorOf(jobManagerProps);

		Future<Object> started = Patterns.ask(jobManager, new Identify(42), deadline.timeLeft().toMillis());

		Await.ready(started, deadline.timeLeft());

		// make the job manager the leader --> this triggers the recovery of all jobs
		myLeaderElectionService.isLeader(leaderSessionID);

		// check that we have successfully recovered the second job
		assertThat(recoveredJobs, containsInAnyOrder(jobId2));
	} finally {
		TestingUtils.stopActor(jobManager);
	}
}
 

import akka.actor.Identify; //导入依赖的package包/类
private void sendIdentifyRequest() {
  getContext().actorSelection(path).tell(new Identify(path), self());
  getContext()
      .system()
      .scheduler()
      .scheduleOnce(Duration.create(3, SECONDS), self(),
          ReceiveTimeout.getInstance(), getContext().dispatcher(), self());
}