### Question:
DefaultSparkJobContext implements SparkJobContext { public static DefaultSparkJobContext create(@Nonnull final ChainableSupplier<SparkJobResponse> responseTask, @Nonnull final SparkJobCacheService cache, @Nonnull final ExecutorService executor) { final String id = UUID.randomUUID().toString(); final Processor<SparkJobStatus, SparkJobStatus> processor = RingBufferProcessor.create(executor, false); final DefaultSparkJobContext context = new DefaultSparkJobContext(id, processor); final ChainableSupplier<SparkJobStatus> statusTask = responseTask .andThen(response -> { context.sparkJobId = response.getId(); return response; }) .andThen(new JobStatusFunction(cache)); final CompletableFuture<SparkJobStatus> future = CompletableFuture.supplyAsync(statusTask, executor) .whenComplete((response, error) -> { if (response != null) { processor.onNext(response); processor.onComplete(); } else if (error != null) { processor.onError(error); } else { processor.onError(new NoSuchElementException()); } }); context.setFuture(future); return context; } private  DefaultSparkJobContext(@Nonnull final String id, @Nonnull final Publisher<SparkJobStatus> publisher); static DefaultSparkJobContext create(@Nonnull final ChainableSupplier<SparkJobResponse> responseTask, @Nonnull final SparkJobCacheService cache, @Nonnull final ExecutorService executor); void cancel(); @Nonnull @Override String getId(); @Nullable String getSparkJobId(); void setFuture(@Nonnull final Future<SparkJobStatus> future); @Override void subscribe(@Nonnull final Subscriber<? super SparkJobStatus> subscriber);  }### Answer:
@Test(expected = CompletionException.class) public void testError() { final ChainableSupplier<SparkJobResponse> task = () -> { throw new UnsupportedOperationException(); }; final DefaultSparkJobContext context = DefaultSparkJobContext.create(task, cache, executor); take(context); }
@Test public void testSuccess() { final SparkJobResponse response = new SparkJobResponse(); response.setId("mock-job"); response.setResult(new SparkJobResult()); response.setStatus(SparkJobResponse.Status.SUCCESS); final DefaultSparkJobContext context = DefaultSparkJobContext.create(() -> response, cache, executor); Assert.assertNotNull(take(context)); Assert.assertEquals(response.getResult(), take(context).getResult()); }