@Test (groups = {"disabledOnTravis"}) public void testExecute() throws Exception { suite.startCluster(); helixManager.connect(); AssertWithBackoff.create().maxSleepMs(1000).backoffFactor(1). assertTrue(ClusterIntegrationTest.isTaskStarted(helixManager, JOB_ID), "Waiting for the job to start..."); String helixJobId = Joiner.on("_").join(JOB_ID, JOB_ID); helixConfig = helixConfig.withValue(GobblinClusterConfigurationKeys.HELIX_INSTANCE_NAME_KEY, ConfigValueFactory.fromAnyRef(IntegrationBasicSuite.WORKER_INSTANCE_0)) .withValue(GobblinClusterConfigurationKeys.HELIX_JOB_ID_KEY, ConfigValueFactory.fromAnyRef(helixJobId)) .withValue(GobblinClusterConfigurationKeys.HELIX_PARTITION_ID_KEY, ConfigValueFactory.fromAnyRef(0)); HelixAssignedParticipantCheck check = new HelixAssignedParticipantCheck(helixConfig); AssertWithBackoff.create().maxSleepMs(100).timeoutMs(2000).backoffFactor(1). assertTrue(ClusterIntegrationTest.isTaskRunning(TASK_STATE_FILE),"Waiting for the task to enter running state"); check.execute(); HelixManager helixManagerOriginal = HelixAssignedParticipantCheck.getHelixManager(); helixManagerOriginal.disconnect(); check.execute(); Assert.assertTrue(HelixAssignedParticipantCheck.getHelixManager() != helixManagerOriginal); helixConfig = helixConfig.withValue(GobblinClusterConfigurationKeys.HELIX_PARTITION_ID_KEY, ConfigValueFactory.fromAnyRef(1)); check = new HelixAssignedParticipantCheck(helixConfig); try { check.execute(); Assert.fail("Expected to throw CommitStepException"); } catch (CommitStepException e) { Assert.assertTrue(e.getClass().equals(CommitStepException.class)); } }
@Override public void execute() throws CommitStepException { log.info(String.format("HelixParticipantCheck step called for Helix Instance: %s, Helix job: %s, Helix partition: %d", this.helixInstanceName, this.helixJob, this.partitionNum)); Callable callable = () -> { JobContext jobContext; try { TaskDriver taskDriver = new TaskDriver(helixManager); jobContext = taskDriver.getJobContext(helixJob); } catch (Exception e) { log.info("Encountered exception when executing " + getClass().getSimpleName(), e); log.info("Refreshing Helix manager.."); refreshHelixManager(); throw e; } if (jobContext != null) { String participant = jobContext.getAssignedParticipant(partitionNum); if (participant != null) { return participant.equalsIgnoreCase(helixInstanceName); } } return false; }; boolean isParticipant; try { isParticipant = retryer.call(callable); } catch (ExecutionException | RetryException e) { log.error("Cannot complete participant assignment check within the retry limit due to: {}", e); isParticipant = true; } if (!isParticipant) { throw new CommitStepException(String.format("Helix instance %s not the assigned participant for partition %d",this.helixInstanceName, this.partitionNum)); } }
HelixAssignedParticipantCheck implements CommitStep { @Override public void execute() throws CommitStepException { log.info(String.format("HelixParticipantCheck step called for Helix Instance: %s, Helix job: %s, Helix partition: %d", this.helixInstanceName, this.helixJob, this.partitionNum)); Callable callable = () -> { JobContext jobContext; try { TaskDriver taskDriver = new TaskDriver(helixManager); jobContext = taskDriver.getJobContext(helixJob); } catch (Exception e) { log.info("Encountered exception when executing " + getClass().getSimpleName(), e); log.info("Refreshing Helix manager.."); refreshHelixManager(); throw e; } if (jobContext != null) { String participant = jobContext.getAssignedParticipant(partitionNum); if (participant != null) { return participant.equalsIgnoreCase(helixInstanceName); } } return false; }; boolean isParticipant; try { isParticipant = retryer.call(callable); } catch (ExecutionException | RetryException e) { log.error("Cannot complete participant assignment check within the retry limit due to: {}", e); isParticipant = true; } if (!isParticipant) { throw new CommitStepException(String.format("Helix instance %s not the assigned participant for partition %d",this.helixInstanceName, this.partitionNum)); } } }
HelixAssignedParticipantCheck implements CommitStep { @Override public void execute() throws CommitStepException { log.info(String.format("HelixParticipantCheck step called for Helix Instance: %s, Helix job: %s, Helix partition: %d", this.helixInstanceName, this.helixJob, this.partitionNum)); Callable callable = () -> { JobContext jobContext; try { TaskDriver taskDriver = new TaskDriver(helixManager); jobContext = taskDriver.getJobContext(helixJob); } catch (Exception e) { log.info("Encountered exception when executing " + getClass().getSimpleName(), e); log.info("Refreshing Helix manager.."); refreshHelixManager(); throw e; } if (jobContext != null) { String participant = jobContext.getAssignedParticipant(partitionNum); if (participant != null) { return participant.equalsIgnoreCase(helixInstanceName); } } return false; }; boolean isParticipant; try { isParticipant = retryer.call(callable); } catch (ExecutionException | RetryException e) { log.error("Cannot complete participant assignment check within the retry limit due to: {}", e); isParticipant = true; } if (!isParticipant) { throw new CommitStepException(String.format("Helix instance %s not the assigned participant for partition %d",this.helixInstanceName, this.partitionNum)); } } HelixAssignedParticipantCheck(Config config); }
HelixAssignedParticipantCheck implements CommitStep { @Override public void execute() throws CommitStepException { log.info(String.format("HelixParticipantCheck step called for Helix Instance: %s, Helix job: %s, Helix partition: %d", this.helixInstanceName, this.helixJob, this.partitionNum)); Callable callable = () -> { JobContext jobContext; try { TaskDriver taskDriver = new TaskDriver(helixManager); jobContext = taskDriver.getJobContext(helixJob); } catch (Exception e) { log.info("Encountered exception when executing " + getClass().getSimpleName(), e); log.info("Refreshing Helix manager.."); refreshHelixManager(); throw e; } if (jobContext != null) { String participant = jobContext.getAssignedParticipant(partitionNum); if (participant != null) { return participant.equalsIgnoreCase(helixInstanceName); } } return false; }; boolean isParticipant; try { isParticipant = retryer.call(callable); } catch (ExecutionException | RetryException e) { log.error("Cannot complete participant assignment check within the retry limit due to: {}", e); isParticipant = true; } if (!isParticipant) { throw new CommitStepException(String.format("Helix instance %s not the assigned participant for partition %d",this.helixInstanceName, this.partitionNum)); } } HelixAssignedParticipantCheck(Config config); static void initHelixManager(Config config); @Override boolean isCompleted(); @Override void execute(); }
HelixAssignedParticipantCheck implements CommitStep { @Override public void execute() throws CommitStepException { log.info(String.format("HelixParticipantCheck step called for Helix Instance: %s, Helix job: %s, Helix partition: %d", this.helixInstanceName, this.helixJob, this.partitionNum)); Callable callable = () -> { JobContext jobContext; try { TaskDriver taskDriver = new TaskDriver(helixManager); jobContext = taskDriver.getJobContext(helixJob); } catch (Exception e) { log.info("Encountered exception when executing " + getClass().getSimpleName(), e); log.info("Refreshing Helix manager.."); refreshHelixManager(); throw e; } if (jobContext != null) { String participant = jobContext.getAssignedParticipant(partitionNum); if (participant != null) { return participant.equalsIgnoreCase(helixInstanceName); } } return false; }; boolean isParticipant; try { isParticipant = retryer.call(callable); } catch (ExecutionException | RetryException e) { log.error("Cannot complete participant assignment check within the retry limit due to: {}", e); isParticipant = true; } if (!isParticipant) { throw new CommitStepException(String.format("Helix instance %s not the assigned participant for partition %d",this.helixInstanceName, this.partitionNum)); } } HelixAssignedParticipantCheck(Config config); static void initHelixManager(Config config); @Override boolean isCompleted(); @Override void execute();  }