FlagFileRepetition extends AbstractRepetition implements Runnable, ProvidesAlias, ProvidesValidations { @Override public void configure(Config config) { super.configure(config); flagFile = new Path(config.getString(FLAG_FILE_LOCATION_CONFIG)); if (config.hasPath(POLL_INTERVAL_CONFIG)) { pollInterval = config.getDuration(POLL_INTERVAL_CONFIG, TimeUnit.MILLISECONDS); } if (config.hasPath(TRIGGER_MODE_CONFIG)) { triggerMode = TriggerMode.valueOf(config.getString(TRIGGER_MODE_CONFIG).toUpperCase()); } if (config.hasPath(NUM_ALLOWED_FAILURES_CONFIG)) { numAllowedFailures = config.getInt(NUM_ALLOWED_FAILURES_CONFIG); } Repetitions.get().submitRegularTask(this, pollInterval); try { fs = FileSystem.get(flagFile.toUri(), Contexts.getSparkSession().sparkContext().hadoopConfiguration()); } catch (IOException e) { throw new RuntimeException("Could not instantiate Hadoop FileSystem"); } }  @Override void configure(Config config); @Override void configureStep(BatchStep step, String name); @Override void run(); @Override String getAlias(); @Override Validations getValidations();  }
@Test public void testRepeatStepPresent() throws IOException { Config config = ConfigUtils.configFromResource("/repetitions/repetitions-flag-config.conf"); File relativeFlagFile = new File("flag"); config = config.withValue("steps.repeater.repetitions.hdfsinnit.file", ConfigValueFactory.fromAnyRef(relativeFlagFile.toURI().toString())); try { BatchStep step = new BatchStep("testFlagRepetition"); ValidationAssert.assertNoValidationFailures(step, config.getConfig("steps.repeater")); step.configure(config.getConfig("steps.repeater")); Set<DataStep> steps = Repetitions.get().getAndClearRepeatingSteps(); assertTrue("Repeating steps should not be populated", steps.isEmpty()); assertTrue(fs.createNewFile(flagFile)); waitForResponse(300, false, 10); waitForResponse(300, true, 1); assertTrue(fs.createNewFile(flagFile)); waitForResponse(300, false, 10); waitForResponse(300, true, 1); } catch (Exception e) { System.err.println(e.getMessage()); fail(); } }
@Test public void testRepeatStepModified() throws IOException { Config config = ConfigUtils.configFromResource("/repetitions/repetitions-flag-config-modified.conf"); File relativeFlagFile = new File("flag"); config = config.withValue("steps.repeater.repetitions.hdfsinnit.file", ConfigValueFactory.fromAnyRef(relativeFlagFile.toURI().toString())); try { BatchStep step = new BatchStep("testFlagRepetitionMod"); step.configure(config.getConfig("steps.repeater")); Set<DataStep> steps = Repetitions.get().getAndClearRepeatingSteps(); assertTrue("Repeating steps should not be populated", steps.isEmpty()); assertTrue(fs.createNewFile(flagFile)); System.out.println("File stat: " + fs.getFileStatus(flagFile)); waitForResponse(300, false, 10); waitForResponse(300, true, 1); long mTime = System.currentTimeMillis() + 1000; System.out.println("Modifying file time to " + mTime); fs.setTimes(flagFile, mTime, mTime); waitForResponse(300, false, 10); waitForResponse(300, true, 1); } catch (Exception e) { System.err.println(e.getMessage()); fail(); } }