com.hazelcast.jet.pipeline.test

Class AssertionSinks

java.lang.Object

com.hazelcast.jet.pipeline.test.AssertionSinks

public final class AssertionSinks
extends Object

Various assertions which can be used to assert items on the output of a pipeline.

In this class there are variants that can be used as sinks in the pipeline. Variants that can be used in-line in a pipeline are in Assertions.

Since:

Jet 3.2

Method Summary

Modifier and Type	Method and Description
static <T> Sink<T>	assertAnyOrder(Collection<? extends T> expected) Asserts that the previous stage emitted the expected items in any order, but nothing else.
static <T> Sink<T>	assertAnyOrder(String message, Collection<? extends T> expected) Asserts that the previous stage emitted the expected items in any order, but nothing else.
static <T> Sink<T>	assertCollected(ConsumerEx<? super List<T>> assertFn) Collects all the received items in a list and once the upstream stage is completed it executes the assertion supplied by assertFn.
static <T> Sink<T>	assertCollectedEventually(int timeoutSeconds, ConsumerEx<? super List<T>> assertFn) Collects all the received items into a list and runs the assertFn every time a new item is received.
static <T> Sink<T>	assertContains(String message, Collection<? extends T> expected) Asserts that the previous stage emitted all of the given items in any order.
static <T> Sink<T>	assertOrdered(Collection<? extends T> expected) Asserts that the previous stage emitted the exact sequence of expected items and nothing else.
static <T> Sink<T>	assertOrdered(String message, Collection<? extends T> expected) Asserts that the previous stage emitted the exact sequence of expected items and nothing else.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

assertOrdered

@Nonnull
public static <T> Sink<T> assertOrdered(@Nullable
                                                 String message,
                                                 @Nonnull
                                                 Collection<? extends T> expected)

Asserts that the previous stage emitted the exact sequence of expected items and nothing else. If the assertion fails, the job will fail with an AssertionError with the given message.

Since Jet jobs are distributed, input from multiple upstream processors is merged in a non-deterministic way. Therefore this assertion is usable only for testing of non-distributed sources.

assertOrdered

@Nonnull
public static <T> Sink<T> assertOrdered(@Nonnull
                                                 Collection<? extends T> expected)

Asserts that the previous stage emitted the exact sequence of expected items and nothing else. If the assertion fails, the job will fail with an AssertionError.

Since Jet jobs are distributed, input from multiple upstream processors is merged in a non-deterministic way. Therefore this assertion is usable only for testing of non-distributed sources.

assertAnyOrder

@Nonnull
public static <T> Sink<T> assertAnyOrder(@Nullable
                                                  String message,
                                                  @Nonnull
                                                  Collection<? extends T> expected)

Asserts that the previous stage emitted the expected items in any order, but nothing else. If the assertion fails, the job will fail with an AssertionError with the given message.

assertAnyOrder

@Nonnull
public static <T> Sink<T> assertAnyOrder(@Nonnull
                                                  Collection<? extends T> expected)

Asserts that the previous stage emitted the expected items in any order, but nothing else. If the assertion fails, the job will fail with an AssertionError.

assertContains

@Nonnull
public static <T> Sink<T> assertContains(@Nullable
                                                  String message,
                                                  @Nonnull
                                                  Collection<? extends T> expected)

Asserts that the previous stage emitted all of the given items in any order. If the assertion fails, the job will fail with a AssertionError with the given message.

assertCollected

@Nonnull
public static <T> Sink<T> assertCollected(@Nonnull
                                                   ConsumerEx<? super List<T>> assertFn)

Collects all the received items in a list and once the upstream stage is completed it executes the assertion supplied by assertFn. If no items were collected, it will be called with empty list.

Not usable in streaming jobs - use assertCollectedEventually(int, com.hazelcast.function.ConsumerEx<? super java.util.List<T>>).

Parameters:

assertFn - assertion to execute once all items are received

assertCollectedEventually

@Nonnull
public static <T> Sink<T> assertCollectedEventually(int timeoutSeconds,
                                                             @Nonnull
                                                             ConsumerEx<? super List<T>> assertFn)

Collects all the received items into a list and runs the assertFn every time a new item is received. An AssertionError thrown from the assertFn will be ignored until timeoutSeconds have passed, after which the last AssertionError will be rethrown. If assertFn throws any other exception, it will be rethrown immediately.

When assertFn completes without any error, the sink will throw an AssertionCompletedException to indicate success. Exception is used to terminate the job so that you can join() it. This also requires that there are no other assertions in the job as this one can complete the job before the other ones succeeded.

The assertion can be validated as follows:

 try {
     jetInstance.newJob(p).join();
     Assert.fail("Job should have completed with an AssertionCompletedException, " +
             "but completed normally");
 } catch (CompletionException e) {
     String errorMsg = e.getCause().getMessage();
     Assert.assertTrue(
             "Job was expected to complete with AssertionCompletedException, but completed with: " + e.getCause(),
             errorMsg.contains(AssertionCompletedException.class.getName())
     );
 }

Parameters:

timeoutSeconds - timeout in seconds, after which any assertion error will be propagated

assertFn - assertion to execute periodically