Class TestSupport
public final class TestSupport extends Object
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 returnstrue
or until the output matches (for streaming sources) - does snapshot or snapshot+restore each time the
complete()
method returnedfalse
and made a progress
init()
and close()
methods of ProcessorSupplier
and ProcessorMetaSupplier
are called if you call
the verifyProcessor(com.hazelcast.function.SupplierEx<com.hazelcast.jet.core.Processor>)
using one of these.
Snapshot & restore
Theoptional
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 containWatermark
s. They will be passed to the
Processor.tryProcessWatermark(com.hazelcast.jet.core.Watermark)
method.
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 toprocess()
. 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 exceedcooperativeTimeout(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"));
- Since:
- 3.0
-
Nested Class Summary
Nested Classes Modifier and Type Class Description static class
TestSupport.TestMode
Describes the current test mode. -
Field Summary
Fields Modifier and Type Field 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 Description void
assertOutput(int outputOrdinalCount, BiConsumer<TestSupport.TestMode,List<List<Object>>> assertFn)
Runs the test with the specified custom assertion.TestSupport
cooperativeTimeout(long timeout)
Iftimeout > 0
, the test will fail if any call to processing method in a cooperative processor exceeds this timeout.TestSupport
disableCompleteCall()
Disables callingcomplete()
method during the test.TestSupport
disableLogging()
Disables logging of input and output objects.TestSupport
disableProgressAssertion()
Disables checking of progress of processing methods (seeclass javadoc
for information on what is "progress").TestSupport
disableSnapshots()
Disable snapshot save and restore before first item and after eachprocess()
andcomplete()
call.TestSupport
executeBeforeEachRun(Runnable runnable)
Execute test before each test runvoid
expectOutput(List<?> expectedOutput)
Sets the expected output and runs the test.void
expectOutputs(List<List<?>> expectedOutputs)
Specifies the expected outputs and runs the test.TestSupport
globalProcessorIndex(int globalProcessorIndex)
Sets the globalProcessorIndex for the ProcessorTestSupport
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 forProcessorMetaSupplier.Context.jetInstance()
TestSupport
localParallelism(int localParallelism)
Sets the localParallelism for the ProcessorTestSupport
localProcessorIndex(int localProcessorIndex)
Sets the localProcessorIndex for the ProcessorTestSupport
outputChecker(BiPredicate<? super List<?>,? super List<?>> outputChecker)
Predicate to compare expected and actual output.TestSupport
runUntilOutputMatches(long timeoutMillis, long extraTimeMillis)
Normally, thecomplete()
method is run repeatedly until it returnstrue
.static Supplier<Processor>
supplierFrom(ProcessorMetaSupplier supplier)
Wraps the providedProcessorMetaSupplier
with aSupplier<Processor>
that returns processors obtained from it.static Supplier<Processor>
supplierFrom(ProcessorMetaSupplier supplier, ProcessorSupplier.Context context)
Wraps the providedProcessorMetaSupplier
with aSupplier<Processor>
that returns processors obtained from it.static Supplier<Processor>
supplierFrom(ProcessorSupplier supplier)
Wraps the providedProcessorSupplier
with aSupplier<Processor>
that returns processors obtained from it.static Supplier<Processor>
supplierFrom(ProcessorSupplier supplier, ProcessorSupplier.Context context)
Wraps the providedProcessorSupplier
with aSupplier<Processor>
that returns processors obtained from it.TestSupport
totalParallelism(int totalParallelism)
Sets the totalParallelism for the Processorstatic TestSupport
verifyProcessor(SupplierEx<Processor> supplier)
static TestSupport
verifyProcessor(ProcessorMetaSupplier supplier)
static TestSupport
verifyProcessor(ProcessorSupplier supplier)
-
Field Details
-
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 Details
-
verifyProcessor
- Parameters:
supplier
- a processor supplier create processor instances
-
verifyProcessor
- Parameters:
supplier
- a processor supplier create processor instances
-
verifyProcessor
- Parameters:
supplier
- a processor supplier create processor instances
-
input
Sets the input objects for processor.The
input
can containWatermark
s; they will be delivered to theProcessor.tryProcessWatermark(com.hazelcast.jet.core.Watermark)
method.Defaults to empty list.
- Returns:
this
instance for fluent API
-
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 ordinalinputs(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
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 ordinalinputs(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
Sets the expected output and runs the test.The
expectedOutput
can containWatermark
s. 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
Specifies the expected outputs and runs the test.The
expectedOutput
can containWatermark
s to assert the watermarks emitted by the processor.- Parameters:
expectedOutputs
- one list for each output ordinal- Throws:
AssertionError
- if some assertion does not hold
-
assertOutput
public void assertOutput(int outputOrdinalCount, BiConsumer<TestSupport.TestMode,List<List<Object>>> assertFn)Runs the test with the specified custom assertion.The consumer takes a list of collected output and the current test mode which can be used in the assertion message.
- Parameters:
outputOrdinalCount
- how many output ordinals should be createdassertFn
- an assertion function which takes the current mode and the collected output
-
disableProgressAssertion
Disables checking of progress of processing methods (seeclass javadoc
for information on what is "progress").- Returns:
this
instance for fluent API
-
runUntilOutputMatches
Normally, thecomplete()
method is run repeatedly until it returnstrue
. This is suitable for processors processing the input or for batch sources. However, if you test a streaming source, thecomplete()
method never returnstrue
. To be able to test such processors, this method allows you to change the behavior to runcomplete()
until the output matches.The
extraTimeMillis
parameter specifies an extra time to callcomplete()
after the output matches. It can be used to ensure that no more items are produced after the output matches.Has no effect if calling
complete()
is disabled.- Parameters:
timeoutMillis
- maximum time to wait for the output to matchextraTimeMillis
- for how long to callcomplete()
after the output matches- Returns:
this
instance for fluent API
-
disableSnapshots
Disable snapshot save and restore before first item and after eachprocess()
andcomplete()
call.- Returns:
this
instance for fluent API
-
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
Disables callingcomplete()
method during the test. Suitable for testing of streaming processors to make sure that the flushing code incomplete()
method is not executed.- Returns:
this
instance for fluent API
-
cooperativeTimeout
Iftimeout > 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
-
localProcessorIndex
Sets the localProcessorIndex for the Processor- Parameters:
localProcessorIndex
- localProcessorIndex, defaults to 0
-
globalProcessorIndex
Sets the globalProcessorIndex for the Processor- Parameters:
globalProcessorIndex
- globalProcessorIndex, default to 0
-
localParallelism
Sets the localParallelism for the Processor- Parameters:
localParallelism
- localParallelism, defaults to 1
-
totalParallelism
Sets the totalParallelism for the Processor- Parameters:
totalParallelism
- totalParallelism, defaults to 1
-
outputChecker
public TestSupport outputChecker(@Nonnull BiPredicate<? super List<?>,? super List<?>> outputChecker)Predicate to compare expected and actual output. Parameters to theBiPredicate
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 useSAME_ITEMS_ANY_ORDER
.- Returns:
this
instance for fluent API
-
jetInstance
Use the given instance forProcessorMetaSupplier.Context.jetInstance()
- Returns:
this
instance for fluent API
-
executeBeforeEachRun
Execute test before each test run- Parameters:
runnable
- runnable to be executed before each test run- Returns:
this
instance for fluent API
-
supplierFrom
Wraps the providedProcessorSupplier
with aSupplier<Processor>
that returns processors obtained from it. -
supplierFrom
public static Supplier<Processor> supplierFrom(ProcessorSupplier supplier, ProcessorSupplier.Context context)Wraps the providedProcessorSupplier
with aSupplier<Processor>
that returns processors obtained from it. -
supplierFrom
Wraps the providedProcessorMetaSupplier
with aSupplier<Processor>
that returns processors obtained from it. -
supplierFrom
public static Supplier<Processor> supplierFrom(ProcessorMetaSupplier supplier, ProcessorSupplier.Context context)Wraps the providedProcessorMetaSupplier
with aSupplier<Processor>
that returns processors obtained from it.
-