@Test public void testSendShutdownRequest() throws Exception { Logger log = LoggerFactory.getLogger("testSendShutdownRequest"); Closer closer = Closer.create(); try { CuratorFramework curatorFramework = TestHelper.createZkClient(this.testingZKServer, closer); final GetInstanceMessageNumFunc getMessageNumFunc = new GetInstanceMessageNumFunc(GobblinClusterManagerTest.class.getSimpleName(), curatorFramework); AssertWithBackoff assertWithBackoff = AssertWithBackoff.create().logger(log).timeoutMs(30000); this.gobblinClusterManager.sendShutdownRequest(); Assert.assertEquals(curatorFramework.checkExists().forPath(String .format("/%s/INSTANCES/%s/MESSAGES", GobblinClusterManagerTest.class.getSimpleName(), TestHelper.TEST_HELIX_INSTANCE_NAME)).getVersion(), 0); assertWithBackoff.assertEquals(getMessageNumFunc, 1, "1 message queued"); assertWithBackoff.assertEquals(getMessageNumFunc, 0, "all messages processed"); } finally { closer.close(); } }
@VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); final int timeout = 300000; GobblinHelixMessagingService messagingService = new GobblinHelixMessagingService(this.multiManager.getJobClusterHelixManager()); int messagesSent = messagingService.send(criteria, shutdownRequest, new NoopReplyHandler(), timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType())); } }
GobblinClusterManager implements ApplicationLauncher, StandardMetricsBridge, LeadershipChangeAwareComponent { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); final int timeout = 300000; GobblinHelixMessagingService messagingService = new GobblinHelixMessagingService(this.multiManager.getJobClusterHelixManager()); int messagesSent = messagingService.send(criteria, shutdownRequest, new NoopReplyHandler(), timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType())); } } }
GobblinClusterManager implements ApplicationLauncher, StandardMetricsBridge, LeadershipChangeAwareComponent { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); final int timeout = 300000; GobblinHelixMessagingService messagingService = new GobblinHelixMessagingService(this.multiManager.getJobClusterHelixManager()); int messagesSent = messagingService.send(criteria, shutdownRequest, new NoopReplyHandler(), timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType())); } } GobblinClusterManager(String clusterName, String applicationId, Config sysConfig,
      Optional<Path> appWorkDirOptional); }
GobblinClusterManager implements ApplicationLauncher, StandardMetricsBridge, LeadershipChangeAwareComponent { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); final int timeout = 300000; GobblinHelixMessagingService messagingService = new GobblinHelixMessagingService(this.multiManager.getJobClusterHelixManager()); int messagesSent = messagingService.send(criteria, shutdownRequest, new NoopReplyHandler(), timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType())); } } GobblinClusterManager(String clusterName, String applicationId, Config sysConfig,
      Optional<Path> appWorkDirOptional); @Override synchronized void start(); @Override synchronized void stop(); @SuppressWarnings("unused") @Subscribe void handleApplicationMasterShutdownRequest(ClusterManagerShutdownRequest shutdownRequest); @Override void close(); @Override Collection<StandardMetrics> getStandardMetricsCollection(); static void main(String[] args); @Override void becomeActive(); @Override void becomeStandby(); }
GobblinClusterManager implements ApplicationLauncher, StandardMetricsBridge, LeadershipChangeAwareComponent { @VisibleForTesting void sendShutdownRequest() { Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.PARTICIPANT); criteria.setSessionSpecific(true); Message shutdownRequest = new Message(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString()); shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString()); shutdownRequest.setMsgState(Message.MessageState.NEW); final int timeout = 300000; GobblinHelixMessagingService messagingService = new GobblinHelixMessagingService(this.multiManager.getJobClusterHelixManager()); int messagesSent = messagingService.send(criteria, shutdownRequest, new NoopReplyHandler(), timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType())); } } GobblinClusterManager(String clusterName, String applicationId, Config sysConfig,
      Optional<Path> appWorkDirOptional); @Override synchronized void start(); @Override synchronized void stop(); @SuppressWarnings("unused") @Subscribe void handleApplicationMasterShutdownRequest(ClusterManagerShutdownRequest shutdownRequest); @Override void close(); @Override Collection<StandardMetrics> getStandardMetricsCollection(); static void main(String[] args); @Override void becomeActive(); @Override void becomeStandby();  }