@Test(dependsOnMethods = "testCreateHelixCluster") public void testSendShutdownRequest() throws Exception { this.helixManager.connect(); this.helixManager.getMessagingService().registerMessageHandlerFactory(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, new TestShutdownMessageHandlerFactory(this)); this.gobblinYarnAppLauncher.connectHelixManager(); this.gobblinYarnAppLauncher.sendShutdownRequest(); Assert.assertEquals(this.curatorFramework.checkExists() .forPath(String.format("/%s/CONTROLLER/MESSAGES", GobblinYarnAppLauncherTest.class.getSimpleName())) .getVersion(), 0); YarnSecurityManagerTest.GetHelixMessageNumFunc getCtrlMessageNum = new YarnSecurityManagerTest.GetHelixMessageNumFunc(GobblinYarnAppLauncherTest.class.getSimpleName(), InstanceType.CONTROLLER, "", this.curatorFramework); AssertWithBackoff assertWithBackoff = AssertWithBackoff.create().logger(LoggerFactory.getLogger("testSendShutdownRequest")).timeoutMs(20000); assertWithBackoff.assertEquals(getCtrlMessageNum, 1, "1 controller message queued"); assertWithBackoff.assertEquals(getCtrlMessageNum, 0, "all controller messages processed"); this.helixManagerManagedHelix.connect(); this.helixManagerManagedHelix.getMessagingService().registerMessageHandlerFactory(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, new TestShutdownMessageHandlerFactory(this)); this.gobblinYarnAppLauncherManagedHelix.connectHelixManager(); this.gobblinYarnAppLauncherManagedHelix.sendShutdownRequest(); Assert.assertEquals(this.curatorFramework.checkExists() .forPath(String.format("/%s/INSTANCES/%s/MESSAGES", this.configManagedHelix.getString(GobblinClusterConfigurationKeys.HELIX_CLUSTER_NAME_KEY), TEST_HELIX_INSTANCE_NAME_MANAGED)) .getVersion(), 0); YarnSecurityManagerTest.GetHelixMessageNumFunc getInstanceMessageNum = new YarnSecurityManagerTest.GetHelixMessageNumFunc(this.configManagedHelix.getString( GobblinClusterConfigurationKeys.HELIX_CLUSTER_NAME_KEY), InstanceType.PARTICIPANT, TEST_HELIX_INSTANCE_NAME_MANAGED, this.curatorFramework); assertWithBackoff = AssertWithBackoff.create().logger(LoggerFactory.getLogger("testSendShutdownRequest")).timeoutMs(20000); assertWithBackoff.assertEquals(getInstanceMessageNum, 1, "1 controller message queued"); assertWithBackoff.assertEquals(getInstanceMessageNum, 0, "all controller messages processed"); }
@VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setResource("%"); if (this.isHelixClusterManaged) { criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setInstanceName(this.helixInstanceName); } else { criteria.setRecipientInstanceType(InstanceType.CONTROLLER); } criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); shutdownRequest.setTgtSessionId("*"); int messagesSent = this.messagingService.send(criteria, shutdownRequest); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } }
GobblinYarnAppLauncher { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setResource("%"); if (this.isHelixClusterManaged) { criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setInstanceName(this.helixInstanceName); } else { criteria.setRecipientInstanceType(InstanceType.CONTROLLER); } criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); shutdownRequest.setTgtSessionId("*"); int messagesSent = this.messagingService.send(criteria, shutdownRequest); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } }
GobblinYarnAppLauncher { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setResource("%"); if (this.isHelixClusterManaged) { criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setInstanceName(this.helixInstanceName); } else { criteria.setRecipientInstanceType(InstanceType.CONTROLLER); } criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); shutdownRequest.setTgtSessionId("*"); int messagesSent = this.messagingService.send(criteria, shutdownRequest); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); }
GobblinYarnAppLauncher { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setResource("%"); if (this.isHelixClusterManaged) { criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setInstanceName(this.helixInstanceName); } else { criteria.setRecipientInstanceType(InstanceType.CONTROLLER); } criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); shutdownRequest.setTgtSessionId("*"); int messagesSent = this.messagingService.send(criteria, shutdownRequest); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); }
GobblinYarnAppLauncher { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setResource("%"); if (this.isHelixClusterManaged) { criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setInstanceName(this.helixInstanceName); } else { criteria.setRecipientInstanceType(InstanceType.CONTROLLER); } criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); shutdownRequest.setTgtSessionId("*"); int messagesSent = this.messagingService.send(criteria, shutdownRequest); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); static final String GOBBLIN_YARN_CONFIG_OUTPUT_PATH; static final String GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String DEFAULT_GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String AZKABAN_APP_LAUNCHER_MODE_KEY; }