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 and Description
static 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 and Description
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 <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 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 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 <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 <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 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 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 <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 <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 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 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	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 Detail

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 Detail

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.