FloRunner { public static <T> Result<T> runTask(Task<T> task, Config config) { return new Result<>(new FloRunner<T>(config).run(task), loadTerminationHooks(config)); } private  FloRunner(Config config); static Result<T> runTask(Task<T> task, Config config); static Result<T> runTask(Task<T> task); static String randomAlphaNumeric(int count);  }
@Test public void valuesCanBeWaitedOn() throws Exception { final String result = runTask(FOO_TASK).value(); assertThat(result, is("foo")); }
@Test public void notReadyExitsTwenty() { final Task<String> task = Task.named("task").ofType(String.class) .process(() -> { throw new NotReady(); }); AtomicInteger status = new AtomicInteger(); runTask(task).waitAndExit(status::set); assertThat(status.get(), is(20)); }
@Test public void notRetriableExitsFifty() { final Task<String> task = Task.named("task").ofType(String.class) .process(() -> { throw new NotRetriable(); }); AtomicInteger status = new AtomicInteger(); runTask(task).waitAndExit(status::set); assertThat(status.get(), is(50)); }
@Test public void exceptionsExitNonZero() { final Task<String> task = Task.named("task").ofType(String.class) .process(() -> { throw new RuntimeException("this task should throw"); }); AtomicInteger status = new AtomicInteger(); runTask(task).waitAndExit(status::set); assertThat(status.get(), is(1)); }
@Test public void ignoreExceptionsFromTerminationHook() { TestTerminationHookFactory.injectHook(exceptionalTerminationHook); AtomicInteger status = new AtomicInteger(); runTask(FOO_TASK).waitAndExit(status::set); verify(exceptionalTerminationHook, times(1)).accept(eq(0)); assertThat(status.get(), is(0)); }
@Test public void validateTerminationHookInvocationOnTaskSuccess() { TestTerminationHookFactory.injectHook(validTerminationHook); AtomicInteger status = new AtomicInteger(); runTask(FOO_TASK).waitAndExit(status::set); verify(validTerminationHook, times(1)).accept(eq(0)); assertThat(status.get(), is(0)); }
@Test public void validateTerminationHookInvocationOnTaskFailure() { final Task<String> task = Task.named("task").ofType(String.class) .process(() -> { throw new RuntimeException("this task should throw"); }); TestTerminationHookFactory.injectHook(validTerminationHook); AtomicInteger status = new AtomicInteger(); runTask(task).waitAndExit(status::set); verify(validTerminationHook, times(1)).accept(eq(1)); assertThat(status.get(), is(1)); }
@Test(expected = RuntimeException.class) public void failOnExceptionalTerminationHookFactory() { TestTerminationHookFactory.injectCreator((config) -> { throw new RuntimeException("factory exception"); }); runTask(FOO_TASK); }
@Test public void taskIdIsInContext() throws Exception { final Task<TaskId> task = Task.named("task").ofType(TaskId.class) .process(() -> Tracing.currentTaskId()); final Result<TaskId> result = runTask(task); assertThat(result.value(), is(task.id())); }
@Test public void tasksRunInProcesses() throws Exception { final Instant today = Instant.now().truncatedTo(ChronoUnit.DAYS); final Instant yesterday = today.minus(1, ChronoUnit.DAYS); final Task<String> baz = Task.named("baz", today).ofType(String.class) .process(() -> { final String bazJvm = jvmName(); log.info("baz: bazJvm={}, today={}", bazJvm, today); return bazJvm; }); final Task<String[]> foo = Task.named("foo", yesterday).ofType(String[].class) .input(() -> baz) .process(bazJvm -> { final String fooJvm = jvmName(); log.info("foo: fooJvm={}, bazJvm={}, yesterday={}", fooJvm, bazJvm, yesterday); return new String[]{bazJvm, fooJvm}; }); final Task<String> quux = Task.named("quux", today).ofType(String.class) .process(() -> { final String quuxJvm = jvmName(); log.info("quux: quuxJvm={}, yesterday={}", quuxJvm, yesterday); return quuxJvm; }); final Task<String[]> bar = Task.named("bar", today, yesterday).ofType(String[].class) .input(() -> foo) .input(() -> quux) .process((bazFooJvms, quuxJvm) -> { final String barJvm = jvmName(); log.info("bar: barJvm={}, bazFooJvms={}, quuxJvm={} today={}, yesterday={}", barJvm, bazFooJvms, quuxJvm, today, yesterday); return Stream.concat( Stream.of(barJvm), Stream.concat( Stream.of(bazFooJvms), Stream.of(quuxJvm)) ).toArray(String[]::new); }); final List<String> jvms = Arrays.asList(runTask(bar).value()); final String mainJvm = jvmName(); log.info("main jvm: {}", mainJvm); log.info("task jvms: {}", jvms); final Set<String> uniqueJvms = new HashSet<>(jvms); assertThat(uniqueJvms.size(), is(4)); assertThat(uniqueJvms, not(contains(mainJvm))); }
@Test public void isTestShouldBeTrueInTestScope() throws Exception { assertThat(FloTesting.isTest(), is(false)); try (TestScope ts = FloTesting.scope()) { assertThat(FloTesting.isTest(), is(true)); final Task<Boolean> isTest = Task.named("task").ofType(Boolean.class) .process(FloTesting::isTest); assertThat(runTask(isTest).future().get(30, SECONDS), is(true)); } assertThat(FloTesting.isTest(), is(false)); }
@Test public void mockingInputsOutputsAndContextShouldBePossibleInTestScope() throws Exception { try (TestScope ts = FloTesting.scope()) { final URI barInput = URI.create("gs: final String jobResult = "42"; final URI publishResult = URI.create("meta: PublishingOutput.mock().publish(jobResult, publishResult); StorageLookup.mock().data("bar", barInput); DataProcessing.mock().result("quux.baz", barInput, jobResult); final Task<String> task = Task.named("task").ofType(String.class) .input(() -> StorageLookup.of("bar")) .output(PublishingOutput.of("foo")) .process((bar, publisher) -> { final String result = DataProcessing.runJob("quux.baz", bar); return publisher.publish(result); }); assertThat(runTask(task).future().get(30, SECONDS), is(jobResult)); assertThat(DataProcessing.mock().jobRuns("quux.baz", barInput), is(1)); assertThat(StorageLookup.mock().lookups("bar"), is(1)); assertThat(PublishingOutput.mock().lookups("foo"), is(1)); assertThat(PublishingOutput.mock().published("foo"), contains(jobResult)); } try (TestScope ts = FloTesting.scope()) { assertThat(PublishingOutput.mock().published("foo"), is(empty())); assertThat(PublishingOutput.mock().lookups("foo"), is(0)); } }
@Test public void mockingContextExistsShouldMakeProcessFnNotRunInTestScope() throws Exception { try (TestScope ts = FloTesting.scope()) { PublishingOutput.mock().value("foo", "17"); final Task<String> task = Task.named("task").ofType(String.class) .output(PublishingOutput.of("foo")) .process(v -> { throw new AssertionError(); }); assertThat(runTask(task).future().get(30, SECONDS), is("17")); assertThat(PublishingOutput.mock().lookups("foo"), is(1)); assertThat(PublishingOutput.mock().published("foo"), is(empty())); } }
@Test public void shouldDryForkInTestMode() throws Exception { final String mainJvmName = jvmName(); final Rock rock = new Rock(); final Task<Rock> task = Task.named("task").ofType(Rock.class) .process(() -> rock); final Rock result; try (TestScope ts = FloTesting.scope()) { result = FloRunner.runTask(task).future().get(30, SECONDS); } assertThat(result, is(not(rock))); assertThat(result.jvmName, is(mainJvmName)); }
@Test public void testOperator() throws Exception { final String mainJvm = jvmName(); final Instant today = Instant.now().truncatedTo(ChronoUnit.DAYS); final Task<JobResult> task = Task.named("task", today).ofType(JobResult.class) .operator(JobOperator.create()) .process(job -> job .options(() -> ImmutableMap.of("quux", 17)) .pipeline(ctx -> ctx.readFrom("foo").map("x + y").writeTo("baz")) .validation(result -> { if (result.records < 5) { throw new AssertionError("Too few records seen!"); } }) .success(result -> new JobResult(jvmName(), "hdfs: final JobResult result = FloRunner.runTask(task) .future().get(30, SECONDS); assertThat(result.jvmName, is(not(mainJvm))); assertThat(result.uri, is("hdfs: }
@Test public void tasksAreObservedByInstrumentedContext() throws Exception { final Task<String> fooTask = Task.named("foo").ofType(String.class) .process(() -> "foo"); final Task<String> barTask = Task.named("bar").ofType(String.class) .operator(JobOperator.create()) .input(() -> fooTask) .process((op, bar) -> op.success(res -> "foo" + bar)); FloRunner.runTask(barTask).future().get(30, SECONDS); RecordingListener.replay(listener); verify(listener).task(argThat(task -> fooTask.id().equals(task.id()))); verify(listener).task(argThat(task -> barTask.id().equals(task.id()))); verify(listener).status(fooTask.id(), Phase.START); verify(listener).status(fooTask.id(), Phase.SUCCESS); verify(listener).status(barTask.id(), Phase.START); verify(listener).status(barTask.id(), Phase.SUCCESS); verify(listener).meta(barTask.id(), Collections.singletonMap("task-id", barTask.id().toString())); verifyNoMoreInteractions(listener); }
@Test public void evaluatesTaskOnce() throws Exception { final String fooRuns = temporaryFolder.newFolder().toString(); Task<String> foo = Task.named("foo").ofType(String.class) .process(() -> { final Path marker = Paths.get(fooRuns, UUID.randomUUID().toString()); try { Files.createFile(marker); } catch (IOException e) { throw new RuntimeException(e); } return marker.toString(); }); Task<String> bar = Task.named("bar").ofType(String.class) .input(() -> foo) .input(() -> foo) .process((foo1, foo2) -> foo1 + foo2); FloRunner.runTask(bar).future().get(); assertThat(Files.list(Paths.get(fooRuns)).count(), is(1L)); }
@Test public void nonBlockingRunnerDoesNotBlock() throws Exception { final Path directory = temporaryFolder.newFolder().toPath(); final File startedFile = directory.resolve("started").toFile(); final File latchFile = directory.resolve("latch").toFile(); final File happenedFile = directory.resolve("happened").toFile(); final Task<Void> task = Task.named("task").ofType(Void.class) .process(() -> { try { startedFile.createNewFile(); while (true) { if (latchFile.exists()) { happenedFile.createNewFile(); return null; } Thread.sleep(100); } } catch (IOException | InterruptedException e) { throw new RuntimeException(e); } }); final Result<Void> result = runTask(task); CompletableFuture.supplyAsync(() -> { while (true) { if (startedFile.exists()) { return true; } try { Thread.sleep(100); } catch (InterruptedException e) { throw new RuntimeException(e); } } }).get(30, SECONDS); try { result.future().get(2, SECONDS); fail(); } catch (TimeoutException ignore) { } assertThat(happenedFile.exists(), is(false)); latchFile.createNewFile(); }
@Test public void blockingRunnerBlocks() throws IOException { final File file = temporaryFolder.newFile(); final Task<Void> task = Task.named("task").ofType(Void.class) .process(() -> { try { Thread.sleep(10); try { Files.write(file.toPath(), "hello".getBytes(UTF_8)); } catch (IOException e) { throw new RuntimeException(e); } } catch (InterruptedException e) { throw new RuntimeException(e); } return null; }); runTask(task).waitAndExit(status -> { }); assertThat(new String(Files.readAllBytes(file.toPath()), UTF_8), is("hello")); }
@Test public void valueIsPassedInFuture() throws Exception { final String result = runTask(FOO_TASK).future().get(30, SECONDS); assertThat(result, is("foo")); }
@Test public void exceptionsArePassed() throws Exception { final Task<String> task = Task.named("foo").ofType(String.class) .process(() -> { throw new RuntimeException("foo"); }); Throwable exception = null; try { runTask(task).value(); } catch (ExecutionException e) { exception = e.getCause(); } assertThat(exception, is(instanceOf(RuntimeException.class))); assertThat(exception.getMessage(), is("foo")); }
@Test public void errorsArePassed() throws Exception { final Task<String> task = Task.named("foo").ofType(String.class) .process(() -> { throw new Error("foo"); }); Throwable exception = null; try { runTask(task).value(); } catch (ExecutionException e) { exception = e.getCause(); } assertThat(exception, is(instanceOf(Error.class))); assertThat(exception.getMessage(), is("foo")); }
@Test public void persistedExitsZero() { final Task<Void> task = Task.named("persisted").ofType(Void.class) .process(() -> { throw new Persisted(); }); AtomicInteger status = new AtomicInteger(1); runTask(task).waitAndExit(status::set); assertThat(status.get(), is(0)); }