@Test(dependsOnMethods = "testGetReconnectableApplicationId") public void testSendShutdownRequest() throws Exception { this.helixManager.connect(); this.helixManager.getMessagingService().registerMessageHandlerFactory(GobblinHelixConstants.SHUTDOWN_MESSAGE_TYPE, new TestShutdownMessageHandlerFactory(this)); this.gobblinAwsClusterLauncher.connectHelixManager(); this.gobblinAwsClusterLauncher.sendShutdownRequest(); Assert.assertEquals(this.curatorFramework.checkExists() .forPath(String.format("/%s/CONTROLLER/MESSAGES", this.helixClusterName)).getVersion(), 0); GetControllerMessageNumFunc getCtrlMessageNum = new GetControllerMessageNumFunc(this.helixClusterName, 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"); }
@VisibleForTesting void sendShutdownRequest() { final Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.CONTROLLER); criteria.setSessionSpecific(true); final 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("*"); final int timeout = 300000; final int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest, shutdownASG(),timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } }
GobblinAWSClusterLauncher { @VisibleForTesting void sendShutdownRequest() { final Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.CONTROLLER); criteria.setSessionSpecific(true); final 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("*"); final int timeout = 300000; final int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest, shutdownASG(),timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } }
GobblinAWSClusterLauncher { @VisibleForTesting void sendShutdownRequest() { final Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.CONTROLLER); criteria.setSessionSpecific(true); final 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("*"); final int timeout = 300000; final int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest, shutdownASG(),timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinAWSClusterLauncher(Config config); }
GobblinAWSClusterLauncher { @VisibleForTesting void sendShutdownRequest() { final Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.CONTROLLER); criteria.setSessionSpecific(true); final 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("*"); final int timeout = 300000; final int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest, shutdownASG(),timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinAWSClusterLauncher(Config config); void launch(); synchronized void stop(); static void main(String[] args); }
GobblinAWSClusterLauncher { @VisibleForTesting void sendShutdownRequest() { final Criteria criteria = new Criteria(); criteria.setInstanceName("%"); criteria.setResource("%"); criteria.setPartition("%"); criteria.setPartitionState("%"); criteria.setRecipientInstanceType(InstanceType.CONTROLLER); criteria.setSessionSpecific(true); final 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("*"); final int timeout = 300000; final int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest, shutdownASG(),timeout); if (messagesSent == 0) { LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType())); } } GobblinAWSClusterLauncher(Config config); void launch(); synchronized void stop(); static void main(String[] args); static final String CLUSTER_NAME_ASG_TAG; static final String CLUSTER_ID_ASG_TAG; static final String ASG_TYPE_ASG_TAG; static final String ASG_TYPE_MASTER; static final String ASG_TYPE_WORKERS; static final String MASTER_ASG_NAME_PREFIX; static final String MASTER_LAUNCH_CONFIG_NAME_PREFIX; static final String WORKERS_ASG_NAME_PREFIX; static final String WORKERS_LAUNCH_CONFIG_PREFIX; }