Package com.hazelcast.jet.core.processor

Class DiagnosticProcessors

java.lang.Object

com.hazelcast.jet.core.processor.DiagnosticProcessors

public final class DiagnosticProcessors
extends Object

Static utility class with factories of sinks and wrappers that log the data flowing through the DAG. These processors are useful while diagnosing the execution of Jet jobs. For other kinds of processors refer to the package-level documentation.

Since:

Jet 3.0

Field Summary

Fields

Modifier and Type	Field	Description
static final FunctionEx<Object,String>	PEEK_DEFAULT_TO_STRING	A function that uses `Object.toString()` for non-arrays, `Arrays.toString()` for arrays of primitive types and `Arrays.deepToString()` for `Object[]`.

Method Summary

Modifier and Type	Method	Description
static <T> SupplierEx<Processor>	peekInputP(FunctionEx<T,? extends CharSequence> toStringFn, PredicateEx<T> shouldLogFn, SupplierEx<Processor> wrapped)	Same as peekInput(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.
static <T> ProcessorMetaSupplier	peekInputP(FunctionEx<T,? extends CharSequence> toStringFn, PredicateEx<T> shouldLogFn, ProcessorMetaSupplier wrapped)	Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates.
static <T> ProcessorSupplier	peekInputP(FunctionEx<T,? extends CharSequence> toStringFn, PredicateEx<T> shouldLogFn, ProcessorSupplier wrapped)	Same as peekInput(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.
static SupplierEx<Processor>	peekInputP(SupplierEx<Processor> wrapped)	Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function.
static ProcessorMetaSupplier	peekInputP(ProcessorMetaSupplier wrapped)	Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function.
static ProcessorSupplier	peekInputP(ProcessorSupplier wrapped)	Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function.
static <T> SupplierEx<Processor>	peekOutputP(FunctionEx<? super T,? extends CharSequence> toStringFn, PredicateEx<? super T> shouldLogFn, SupplierEx<Processor> wrapped)	Same as peekOutput(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.
static <T> ProcessorMetaSupplier	peekOutputP(FunctionEx<? super T,? extends CharSequence> toStringFn, PredicateEx<? super T> shouldLogFn, ProcessorMetaSupplier wrapped)	Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates.
static <T> ProcessorSupplier	peekOutputP(FunctionEx<? super T,? extends CharSequence> toStringFn, PredicateEx<? super T> shouldLogFn, ProcessorSupplier wrapped)	Same as peekOutput(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.
static SupplierEx<Processor>	peekOutputP(SupplierEx<Processor> wrapped)	Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static ProcessorMetaSupplier	peekOutputP(ProcessorMetaSupplier wrapped)	Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static ProcessorSupplier	peekOutputP(ProcessorSupplier wrapped)	Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static <K, V> SupplierEx<Processor>	peekSnapshotP(FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn, PredicateEx<? super Map.Entry<K,V>> shouldLogFn, SupplierEx<Processor> wrapped)	Same as peekSnapshot(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.
static <K, V> ProcessorMetaSupplier	peekSnapshotP(FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn, PredicateEx<? super Map.Entry<K,V>> shouldLogFn, ProcessorMetaSupplier wrapped)	Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates.
static <K, V> ProcessorSupplier	peekSnapshotP(FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn, PredicateEx<? super Map.Entry<K,V>> shouldLogFn, ProcessorSupplier wrapped)	Same as peekSnapshot(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.
static SupplierEx<Processor>	peekSnapshotP(SupplierEx<Processor> wrapped)	Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static ProcessorMetaSupplier	peekSnapshotP(ProcessorMetaSupplier wrapped)	Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static ProcessorSupplier	peekSnapshotP(ProcessorSupplier wrapped)	Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.
static ProcessorMetaSupplier	writeLoggerP()	Convenience for writeLoggerP(FunctionEx) that uses toString() as toStringFn.
static <T> ProcessorMetaSupplier	writeLoggerP(FunctionEx<T,? extends CharSequence> toStringFn)	Returns a meta-supplier of processors for a sink vertex that logs all the data items it receives.

Methods inherited from class java.lang.Object

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

Field Details

PEEK_DEFAULT_TO_STRING

public static final FunctionEx<Object,String> PEEK_DEFAULT_TO_STRING

A function that uses `Object.toString()` for non-arrays, `Arrays.toString()` for arrays of primitive types and `Arrays.deepToString()` for `Object[]`. Used for `peek()` function in the DAG and Pipeline API.

Since:

5.1

Method Details

writeLoggerP

@Nonnull
public static <T> ProcessorMetaSupplier writeLoggerP(@Nonnull
 FunctionEx<T,? extends CharSequence> toStringFn)

Returns a meta-supplier of processors for a sink vertex that logs all the data items it receives. The log category is com.hazelcast.jet.impl.connector.WriteLoggerP.<vertexName>#<processorIndex> and the level is INFO. Watermark items are always logged, but at FINE level; they are not passed to toStringFn.

The vertex logs each item on whichever cluster member it happens to receive it. Its primary purpose is for development, when running Jet on a local machine.

Type Parameters:

T - stream item type

Parameters:

toStringFn - a function that returns a string representation of a stream item

writeLoggerP

@Nonnull
public static ProcessorMetaSupplier writeLoggerP()

Convenience for writeLoggerP(FunctionEx) that uses toString() as toStringFn.

peekInputP

@Nonnull
public static <T> ProcessorMetaSupplier peekInputP(@Nonnull
 FunctionEx<T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<T> shouldLogFn,
 @Nonnull
 ProcessorMetaSupplier wrapped)

Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates. For each item the wrapped processor removes from the inbox, the wrapping processor:

1. uses the shouldLogFn predicate to see whether to log the item
2. if the item passed, uses toStringFn to get a string representation of the item
3. logs the string at the INFO level, the logger is com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>. The text is prefixed with "Input from X: ", where X is the edge ordinal the item is received from. Received watermarks are prefixed with just "Input: ".

Note: Watermarks are always logged. Watermark objects are not passed to shouldLogFn and toStringFn.

Type Parameters:

T - input item type

Parameters:

toStringFn - a function that returns the string representation of the item. You can use PEEK_DEFAULT_TO_STRING.

shouldLogFn - a function to filter the logged items. You can use alwaysTrue() as a pass-through filter when you don't need any filtering.

wrapped - The wrapped meta-supplier.

See Also:

• peekOutputP(FunctionEx, PredicateEx, ProcessorMetaSupplier)
• peekSnapshotP(FunctionEx, PredicateEx, ProcessorMetaSupplier)

peekInputP

@Nonnull
public static <T> ProcessorSupplier peekInputP(@Nonnull
 FunctionEx<T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<T> shouldLogFn,
 @Nonnull
 ProcessorSupplier wrapped)

Same as peekInput(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.

peekInputP

@Nonnull
public static <T> SupplierEx<Processor> peekInputP(@Nonnull
 FunctionEx<T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<T> shouldLogFn,
 @Nonnull
 SupplierEx<Processor> wrapped)

Same as peekInput(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.

peekInputP

@Nonnull
public static ProcessorMetaSupplier peekInputP(@Nonnull
 ProcessorMetaSupplier wrapped)

Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function.

peekInputP

@Nonnull
public static ProcessorSupplier peekInputP(@Nonnull
 ProcessorSupplier wrapped)

Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function. This variant accepts a ProcessorSupplier instead of a meta-supplier.

peekInputP

@Nonnull
public static SupplierEx<Processor> peekInputP(@Nonnull
 SupplierEx<Processor> wrapped)

Convenience for peekInput(toStringFn, shouldLogFn, metaSupplier) with a pass-through filter and Object#toString as the formatting function. This variant accepts a SupplierEx of processors instead of a meta-supplier.

peekOutputP

@Nonnull
public static <T> ProcessorMetaSupplier peekOutputP(@Nonnull
 FunctionEx<? super T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super T> shouldLogFn,
 @Nonnull
 ProcessorMetaSupplier wrapped)

Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates. For each item the wrapped processor adds to the outbox, the wrapping processor:

1. uses the shouldLogFn predicate to see whether to log the item
2. if the item passed, uses toStringFn to get a string representation of the item
3. logs the string at the INFO level, the logger is com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>. The logged text is prefixed with "Output to X: ", where X is the edge ordinal the item is sent to

Technically speaking, snapshot data is emitted to the same outbox as regular data, but this wrapper only logs the regular data. See peekSnapshot().

Logging of Watermarks

There are two kinds of watermarks:

1. Watermarks originated in the processor, prefixed in the logs with "Output to N: "
2. Watermarks received on input, which are forwarded automatically. These are prefixed with "Output forwarded: "

Both are always logged. Watermark objects are not passed to shouldLogFn or toStringFn.

Type Parameters:

T - output item type

Parameters:

toStringFn - a function that returns the string representation of the item. You can use PEEK_DEFAULT_TO_STRING.

shouldLogFn - a function to filter the logged items. You can use alwaysTrue() as a pass-through filter when you don't need any filtering.

wrapped - The wrapped meta-supplier.

See Also:

• peekInputP(FunctionEx, PredicateEx, ProcessorMetaSupplier)
• peekSnapshotP(FunctionEx, PredicateEx, ProcessorMetaSupplier)

peekOutputP

@Nonnull
public static <T> ProcessorSupplier peekOutputP(@Nonnull
 FunctionEx<? super T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super T> shouldLogFn,
 @Nonnull
 ProcessorSupplier wrapped)

Same as peekOutput(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.

peekOutputP

@Nonnull
public static <T> SupplierEx<Processor> peekOutputP(@Nonnull
 FunctionEx<? super T,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super T> shouldLogFn,
 @Nonnull
 SupplierEx<Processor> wrapped)

Same as peekOutput(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.

peekOutputP

@Nonnull
public static ProcessorMetaSupplier peekOutputP(@Nonnull
 ProcessorMetaSupplier wrapped)

Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.

peekOutputP

@Nonnull
public static ProcessorSupplier peekOutputP(@Nonnull
 ProcessorSupplier wrapped)

Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function. This variant accepts a ProcessorSupplier instead of a meta-supplier.

peekOutputP

@Nonnull
public static SupplierEx<Processor> peekOutputP(@Nonnull
 SupplierEx<Processor> wrapped)

Convenience for peekOutput(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function. This variant accepts a SupplierEx of processors instead of a meta-supplier.

peekSnapshotP

@Nonnull
public static <K,
V> ProcessorMetaSupplier peekSnapshotP(@Nonnull
 FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super Map.Entry<K,V>> shouldLogFn,
 @Nonnull
 ProcessorMetaSupplier wrapped)

Returns a meta-supplier that wraps the provided one and adds a logging layer to each processor it creates. For each item the wrapped processor adds to the snapshot storage, the wrapping processor:

1. uses the shouldLogFn predicate to see whether to log the item
2. if the item passed, uses toStringFn to get a string representation of the item
3. logs the string at the INFO level, the category being com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>

Type Parameters:

K - type of the key emitted to the snapshot

V - type of the value emitted to the snapshot

Parameters:

toStringFn - a function that returns the string representation of the item. You can use DEFAULT_TO_STRING

shouldLogFn - a function to filter the logged items. You can use alwaysTrue() as a pass-through filter when you don't need any filtering.

wrapped - The wrapped meta-supplier.

See Also:

• peekInputP(FunctionEx, PredicateEx, ProcessorMetaSupplier)
• peekOutputP(FunctionEx, PredicateEx, ProcessorMetaSupplier)

peekSnapshotP

@Nonnull
public static <K,
V> ProcessorSupplier peekSnapshotP(@Nonnull
 FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super Map.Entry<K,V>> shouldLogFn,
 @Nonnull
 ProcessorSupplier wrapped)

Same as peekSnapshot(toStringFn, shouldLogFn, metaSupplier), but accepts a ProcessorSupplier instead of a meta-supplier.

peekSnapshotP

@Nonnull
public static <K,
V> SupplierEx<Processor> peekSnapshotP(@Nonnull
 FunctionEx<? super Map.Entry<K,V>,? extends CharSequence> toStringFn,
 @Nonnull
 PredicateEx<? super Map.Entry<K,V>> shouldLogFn,
 @Nonnull
 SupplierEx<Processor> wrapped)

Same as peekSnapshot(toStringFn, shouldLogFn, metaSupplier), but accepts a SupplierEx of processors instead of a meta-supplier.

peekSnapshotP

@Nonnull
public static SupplierEx<Processor> peekSnapshotP(@Nonnull
 SupplierEx<Processor> wrapped)

Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function. This variant accepts a SupplierEx of processors instead of a meta-supplier.

peekSnapshotP

@Nonnull
public static ProcessorMetaSupplier peekSnapshotP(@Nonnull
 ProcessorMetaSupplier wrapped)

Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function.

peekSnapshotP

@Nonnull
public static ProcessorSupplier peekSnapshotP(@Nonnull
 ProcessorSupplier wrapped)

Convenience for peekSnapshot(toStringFn, shouldLogFn, metaSupplier with a pass-through filter and Object#toString as the formatting function. This variant accepts a ProcessorSupplier instead of a meta-supplier.