HandlerRedirectUtils { public static String getRedirectAddress( String localJobManagerAddress, Tuple2<ActorGateway, Integer> leader) throws Exception { final String leaderAddress = leader._1().path(); final int webMonitorPort = leader._2(); final String jobManagerName = localJobManagerAddress.substring(localJobManagerAddress.lastIndexOf("/") + 1); if (!localJobManagerAddress.equals(leaderAddress) && !leaderAddress.equals(JobManager.getLocalJobManagerAkkaURL(Option.apply(jobManagerName)))) { Matcher matcher = LeaderAddressHostPattern.matcher(leaderAddress); if (matcher.matches()) { String redirectAddress = String.format("%s:%d", matcher.group(1), webMonitorPort); return redirectAddress; } else { LOG.warn("Unexpected leader address pattern {}. Cannot extract host.", leaderAddress); } } return null; }  static String getRedirectAddress(
			String localJobManagerAddress,
			Tuple2<ActorGateway, Integer> leader); static HttpResponse getRedirectResponse(String redirectAddress, String path); static HttpResponse getUnavailableResponse();  }
@Test public void testGetRedirectAddressWithLocalAkkaPath() throws Exception { ActorGateway leaderGateway = new DummyActorGateway("akka: Tuple2<ActorGateway, Integer> leader = new Tuple2<>(leaderGateway, 1235); String redirectingAddress =HandlerRedirectUtils.getRedirectAddress(localJobManagerAddress, leader); Assert.assertNull(redirectingAddress); }
@Test public void testGetRedirectAddressWithRemoteAkkaPath() throws Exception { ActorGateway leaderGateway = new DummyActorGateway(remotePath); Tuple2<ActorGateway, Integer> leader = new Tuple2<>(leaderGateway, 1235); String redirectingAddress =HandlerRedirectUtils.getRedirectAddress(localJobManagerAddress, leader); Assert.assertEquals(remoteURL, redirectingAddress); }