com.hazelcast.jet.core.test

Class TestSupport

java.lang.Object

com.hazelcast.jet.core.test.TestSupport

public final class TestSupport
extends Object

A utility to test processors. It will initialize the processor instance, pass input items to it and assert the outbox contents.

The test process does the following:

• initializes the processor by calling Processor.init(com.hazelcast.jet.core.Outbox, com.hazelcast.jet.core.Processor.Context)
• does snapshot or snapshot+restore (optional, see below)
• calls Processor.process(int, com.hazelcast.jet.core.Inbox), in two scenarios:
  • the inbox contains one input item
  • the inbox contains all input items (if snapshots are not restored)
• every time the inbox gets empty does snapshot or snapshot+restore
• optionally calls Processor.complete() until it returns true or calls it until specified timeout elapses (for streaming sources)
• does snapshot or snapshot+restore each time the complete() method returned false and made a progress

The init() and complete() methods of ProcessorSupplier and ProcessorMetaSupplier are called if you call the verifyProcessor(com.hazelcast.jet.function.DistributedSupplier<com.hazelcast.jet.core.Processor>) using one of these.

Snapshot & restore

The optional snapshot+restore test procedure:

• saveToSnapshot() is called. If we are not doing restore, this is the last step.
• new processor instance is created, from now on only this instance will be used
• snapshot is restored using restoreFromSnapshot()
• finishSnapshotRestore() is called

Watermark handling

The input can contain Watermarks. They will be passed to the Processor.tryProcessWatermark(com.hazelcast.jet.core.Watermark) method and can be asserted in the output.

Progress assertion

For each call to any processing method the progress is asserted (optional). The processor must do at least one of these:

• take something from inbox
• put something to outbox
• for boolean-returning methods, returning true is considered as making progress

Outbox rejection

A 1-capacity outbox will be provided, which will additionally be full in every other call to process(). This will test the edge case: the process() method is called even when the outbox is full to give the processor a chance to process the inbox. The snapshot bucket will also have capacity of 1.

Cooperative processors

For cooperative processors, time spent in each call to processing method must not exceed cooperativeTimeout(long).

Non-covered cases

This class does not cover these cases:

• Checking that the state of a stateful processor is empty at the end (you can do that yourself afterwards with the last instance returned from your supplier).
• This utility never calls Processor.tryProcess().

Example usage

This will test one of the jet-provided processors:

 TestSupport.verifyProcessor(Processors.map((String s) -> s.toUpperCase()))
            .disableCompleteCall()             // enabled by default
            .disableLogging()                  // enabled by default
            .disableProgressAssertion()        // enabled by default
            .disableSnapshots()                // enabled by default
            .cooperativeTimeout(<timeoutInMs>) // default is 1000
            .outputChecker(<function>)         // default is `Objects::equal`
            .input(asList("foo", "bar"))       // default is `emptyList()`
            .expectOutput(asList("FOO", "BAR"));
 

Field Summary

Fields

Modifier and Type	Field and Description
static BiPredicate<List<?>,List<?>>	SAME_ITEMS_ANY_ORDER An output checker that will claim actual and expected object lists as equal if they both contain the same items, in any order.

Method Summary

Modifier and Type	Method and Description
TestSupport	cooperativeTimeout(long timeout) If timeout > 0, the test will fail if any call to processing method in a cooperative processor exceeds this timeout.
TestSupport	disableCompleteCall() Disables calling complete() method during the test.
TestSupport	disableLogging() Disables logging of input and output objects.
TestSupport	disableProgressAssertion() Disables checking of progress of processing methods (see class javadoc for information on what is "progress").
TestSupport	disableRunUntilCompleted(long timeoutMillis) Normally, the complete() method is run repeatedly until it returns true.
TestSupport	disableSnapshots() Disable snapshot save and restore before first item and after each process() and complete() call.
void	expectOutput(List<?> expectedOutput) Sets the expected output and runs the test.
void	expectOutputs(List<List<?>> expectedOutputs) Sets the expected outputs and runs the test.
TestSupport	input(List<?> input) Sets the input objects for processor.
TestSupport	inputs(List<List<?>> inputs) Sets the input objects for the processor on multiple input ordinals.
TestSupport	inputs(List<List<?>> inputs, int[] priorities) Sets the input objects for the processor on multiple input ordinals.
TestSupport	jetInstance(JetInstance jetInstance) Use the given instance for ProcessorMetaSupplier.Context.jetInstance()
static String	listToString(List<?> list) Converts a list to a string putting toString() of each element on separate line.
TestSupport	outputChecker(BiPredicate<? super List<?>,? super List<?>> outputChecker) Predicate to compare expected and actual output.
static Supplier<Processor>	supplierFrom(ProcessorMetaSupplier supplier) Wraps the provided ProcessorMetaSupplier with a Supplier<Processor> that returns processors obtained from it.
static Supplier<Processor>	supplierFrom(ProcessorSupplier supplier) Wraps the provided ProcessorSupplier with a Supplier<Processor> that returns processors obtained from it.
static TestSupport	verifyProcessor(DistributedSupplier<Processor> supplier)
static TestSupport	verifyProcessor(ProcessorMetaSupplier supplier)
static TestSupport	verifyProcessor(ProcessorSupplier supplier)

Methods inherited from class java.lang.Object

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

Field Detail

SAME_ITEMS_ANY_ORDER

public static final BiPredicate<List<?>,List<?>> SAME_ITEMS_ANY_ORDER

An output checker that will claim actual and expected object lists as equal if they both contain the same items, in any order. If some item is expected multiple times, it must also be present the same number of times in the actual output.

Use as an argument for outputChecker(BiPredicate).

Method Detail

verifyProcessor

public static TestSupport verifyProcessor(@Nonnull
                                          DistributedSupplier<Processor> supplier)

Parameters:

supplier - a processor supplier create processor instances

verifyProcessor

public static TestSupport verifyProcessor(@Nonnull
                                          ProcessorSupplier supplier)

Parameters:

supplier - a processor supplier create processor instances

verifyProcessor

public static TestSupport verifyProcessor(@Nonnull
                                          ProcessorMetaSupplier supplier)

Parameters:

supplier - a processor supplier create processor instances

input

public TestSupport input(@Nonnull
                         List<?> input)

Sets the input objects for processor.

The input can contain Watermarks; they will be delivered to the Processor.tryProcessWatermark(com.hazelcast.jet.core.Watermark) method.

Defaults to empty list.

Returns:

this instance for fluent API

inputs

public TestSupport inputs(@Nonnull
                          List<List<?>> inputs)

Sets the input objects for the processor on multiple input ordinals. Items will be passed to the processor in round-robin fashion: item0 from input0, item0 from input1, item1 from input0 etc.

See also:

• input(List) - if you have just one input ordinal
• inputs(List, int[]) - if you want to specify input priorities

Parameters:

inputs - one list of input items for each input edge

Returns:

this instance for fluent API

inputs

public TestSupport inputs(@Nonnull
                          List<List<?>> inputs,
                          int[] priorities)

Sets the input objects for the processor on multiple input ordinals. Items will be passed to the processor according to priority: lower is higher. If two inputs have the same priority, they will be passed in round-robin fashion.

See also:

• input(List) - if you have just one input ordinal
• inputs(List) - if all inputs are of equal priority

Parameters:

inputs - one list of input items for each input edge

Returns:

this instance for fluent API

expectOutput

public void expectOutput(@Nonnull
                         List<?> expectedOutput)

Sets the expected output and runs the test.

The expectedOutput can contain Watermarks. Each Watermark in the input will be found in the output, as well as other watermarks the processor emits.

Throws:

AssertionError - If some assertion does not hold.

expectOutputs

public void expectOutputs(@Nonnull
                          List<List<?>> expectedOutputs)

Sets the expected outputs and runs the test.

The expectedOutput can contain Watermarks. Each Watermark in the input will be found in the output, as well as other watermarks the processor emits.

Parameters:

expectedOutputs - one list for each out output ordinal

Throws:

AssertionError - if some assertion does not hold

disableProgressAssertion

public TestSupport disableProgressAssertion()

Disables checking of progress of processing methods (see class javadoc for information on what is "progress").

Returns:

this instance for fluent API

disableRunUntilCompleted

public TestSupport disableRunUntilCompleted(long timeoutMillis)

Normally, the complete() method is run repeatedly until it returns true. But in infinite source processors the method never returns true. To be able test such processors, this method allows you to disable the "run until completed" behavior and instead run the complete() for a specified time.

If the timeout > 0, the complete() method is called repeatedly until the timeout elapses. After that, the output is compared using the output checker. The complete() method is also not allowed to return true in this case.

If the timeout is <= 0 (the default), complete() method is called until it returns true, after which the output is checked.

Has no effect if complete() call is disabled.

Parameters:

timeoutMillis - how long to wait until outputs match

Returns:

this instance for fluent API

disableSnapshots

public TestSupport disableSnapshots()

Disable snapshot save and restore before first item and after each process() and complete() call.

Returns:

this instance for fluent API

disableLogging

public TestSupport disableLogging()

Disables logging of input and output objects. Normally they are logged as they are processed to standard output.

Returns:

this instance for fluent API

disableCompleteCall

public TestSupport disableCompleteCall()

Disables calling complete() method during the test. Suitable for testing of streaming processors to make sure that the flushing code in complete() method is not executed.

Returns:

this instance for fluent API

cooperativeTimeout

public TestSupport cooperativeTimeout(long timeout)

If timeout > 0, the test will fail if any call to processing method in a cooperative processor exceeds this timeout. Has no effect for non-cooperative processors.

Default value is COOPERATIVE_TIME_LIMIT_MS_FAIL ms. Useful to set to 0 during debugging.

Returns:

this instance for fluent API

outputChecker

public TestSupport outputChecker(@Nonnull
                                 BiPredicate<? super List<?>,? super List<?>> outputChecker)

Predicate to compare expected and actual output. Parameters to the BiPredicate are the list of expected items and the list of actual processor output.

Defaults to Objects::equals, which will pass, if both lists contain equal objects in the same order. If the ordering doesn't matter, you can use SAME_ITEMS_ANY_ORDER.

Returns:

this instance for fluent API

jetInstance

public TestSupport jetInstance(@Nonnull
                               JetInstance jetInstance)

Use the given instance for ProcessorMetaSupplier.Context.jetInstance()

Returns:

this instance for fluent API

supplierFrom

public static Supplier<Processor> supplierFrom(ProcessorSupplier supplier)

Wraps the provided ProcessorSupplier with a Supplier<Processor> that returns processors obtained from it.

supplierFrom

public static Supplier<Processor> supplierFrom(ProcessorMetaSupplier supplier)

Wraps the provided ProcessorMetaSupplier with a Supplier<Processor> that returns processors obtained from it.

listToString

public static String listToString(List<?> list)

Converts a list to a string putting toString() of each element on separate line. It is useful to transform list inputs to assertEquals(): the exception will show the entire collections instead of just non-equal sizes or the first non-equal element.

Parameters:

list - Input list

Returns:

Output string