com.hazelcast.jet.pipeline

## Interface StreamStage<T>

- Type Parameters:
    `T` - the type of items coming out of this stage
  All Superinterfaces:
    GeneralStage<T>, Stage

      public interface StreamStage<T>
      extends GeneralStage<T>

  A stage in a distributed computation `pipeline` that will observe an unbounded amount of data (i.e., an event stream). It accepts input from its upstream stages (if any) and passes its output to its downstream stages.
  Since:
    3.0

- - ### Method Summary

    All Methods Instance Methods Abstract Methods Default Methods
    Modifier and Type | Method and Description
    --- | ---
    `default <R> StreamStage<R>` | `apply(FunctionEx<? super StreamStage<T>,? extends StreamStage<R>> transformFn)` Transforms `this` stage using the provided `transformFn` and returns the transformed stage.
    `<R> StreamStage<R>` | `customTransform(String stageName, ProcessorMetaSupplier procSupplier)` Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.
    `default <R> StreamStage<R>` | `customTransform(String stageName, ProcessorSupplier procSupplier)` Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.
    `default <R> StreamStage<R>` | `customTransform(String stageName, SupplierEx<Processor> procSupplier)` Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.
    `StreamStage<T>` | `filter(PredicateEx<T> filterFn)` Attaches a filtering stage which applies the provided predicate function to each input item to decide whether to pass the item to the output or to discard it.
    `<S> StreamStage<T>` | `filterStateful(SupplierEx<? extends S> createFn, BiPredicateEx<? super S,? super T> filterFn)` Attaches a stage that performs a stateful filtering operation.
    `<C> StreamStage<T>` | `filterUsingContext(ContextFactory<C> contextFactory, BiPredicateEx<? super C,? super T> filterFn)` Attaches a filtering stage which applies the provided predicate function to each input item to decide whether to pass the item to the output or to discard it.
    `<C> StreamStage<T>` | `filterUsingContextAsync(ContextFactory<C> contextFactory, BiFunctionEx<? super C,? super T,? extends CompletableFuture<Boolean>> filterAsyncFn)` Asynchronous version of `GeneralStage.filterUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiPredicateEx<? super C, ? super T>)`: the `filterAsyncFn` returns a `CompletableFuture<Boolean>` instead of just a `boolean`.
    `<R> StreamStage<R>` | `flatMap(FunctionEx<? super T,? extends Traverser<? extends R>> flatMapFn)` Attaches a flat-mapping stage which applies the supplied function to each input item independently and emits all the items from the `Traverser` it returns.
    `<S,R> StreamStage<R>` | `flatMapStateful(SupplierEx<? extends S> createFn, BiFunctionEx<? super S,? super T,? extends Traverser<R>> flatMapFn)` Attaches a stage that performs a stateful flat-mapping operation.
    `<C,R> StreamStage<R>` | `flatMapUsingContext(ContextFactory<C> contextFactory, BiFunctionEx<? super C,? super T,? extends Traverser<R>> flatMapFn)` Attaches a flat-mapping stage which applies the supplied function to each input item independently and emits all items from the `Traverser` it returns as the output items.
    `<C,R> StreamStage<R>` | `flatMapUsingContextAsync(ContextFactory<C> contextFactory, BiFunctionEx<? super C,? super T,? extends CompletableFuture<Traverser<R>>> flatMapAsyncFn)` Asynchronous version of `GeneralStage.flatMapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiFunctionEx<? super C, ? super T, ? extends com.hazelcast.jet.Traverser<R>>)`: the `flatMapAsyncFn` returns a `CompletableFuture<Traverser<R>>` instead of just `Traverser<R>`.
    `<K> StreamStageWithKey<T,K>` | `groupingKey(FunctionEx<? super T,? extends K> keyFn)` Specifies the function that will extract a key from the items in the associated pipeline stage.
    `<K,T1_IN,T1,R> StreamStage<R>` | `hashJoin(BatchStage<T1_IN> stage1, JoinClause<K,? super T,? super T1_IN,? extends T1> joinClause1, BiFunctionEx<T,T1,R> mapToOutputFn)` Attaches to both this and the supplied stage a hash-joining stage and returns it.
    `<K1,K2,T1_IN,T2_IN,T1,T2,R> StreamStage<R>` | `hashJoin2(BatchStage<T1_IN> stage1, JoinClause<K1,? super T,? super T1_IN,? extends T1> joinClause1, BatchStage<T2_IN> stage2, JoinClause<K2,? super T,? super T2_IN,? extends T2> joinClause2, TriFunction<T,T1,T2,R> mapToOutputFn)` Attaches to this and the two supplied stages a hash-joining stage and returns it.
    `default StreamHashJoinBuilder<T>` | `hashJoinBuilder()` Returns a fluent API builder object to construct a hash join operation with any number of contributing stages.
    `<R> StreamStage<R>` | `map(FunctionEx<? super T,? extends R> mapFn)` Attaches a mapping stage which applies the given function to each input item independently and emits the function's result as the output item.
    `<S,R> StreamStage<R>` | `mapStateful(SupplierEx<? extends S> createFn, BiFunctionEx<? super S,? super T,? extends R> mapFn)` Attaches a stage that performs a stateful mapping operation.
    `<C,R> StreamStage<R>` | `mapUsingContext(ContextFactory<C> contextFactory, BiFunctionEx<? super C,? super T,? extends R> mapFn)` Attaches a mapping stage which applies the supplied function to each input item independently and emits the function's result as the output item.
    `<C,R> StreamStage<R>` | `mapUsingContextAsync(ContextFactory<C> contextFactory, BiFunctionEx<? super C,? super T,? extends CompletableFuture<R>> mapAsyncFn)` Asynchronous version of `GeneralStage.mapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiFunctionEx<? super C, ? super T, ? extends R>)`: the `mapAsyncFn` returns a `CompletableFuture<R>` instead of just `R`.
    `default <K,V,R> StreamStage<R>` | `mapUsingIMap(IMap<K,V> iMap, FunctionEx<? super T,? extends K> lookupKeyFn, BiFunctionEx<? super T,? super V,? extends R> mapFn)` Attaches a mapping stage where for each item a lookup in the supplied `IMap` is performed and the result of the lookup is merged with the item and emitted.
    `default <K,V,R> StreamStage<R>` | `mapUsingIMap(String mapName, FunctionEx<? super T,? extends K> lookupKeyFn, BiFunctionEx<? super T,? super V,? extends R> mapFn)` Attaches a mapping stage where for each item a lookup in the `IMap` with the supplied name is performed and the result of the lookup is merged with the item and emitted.
    `default <K,V,R> StreamStage<R>` | `mapUsingReplicatedMap(ReplicatedMap<K,V> replicatedMap, FunctionEx<? super T,? extends K> lookupKeyFn, BiFunctionEx<? super T,? super V,? extends R> mapFn)` Attaches a mapping stage where for each item a lookup in the supplied `ReplicatedMap` is performed and the result of the lookup is merged with the item and emitted.
    `default <K,V,R> StreamStage<R>` | `mapUsingReplicatedMap(String mapName, FunctionEx<? super T,? extends K> lookupKeyFn, BiFunctionEx<? super T,? super V,? extends R> mapFn)` Attaches a mapping stage where for each item a lookup in the `ReplicatedMap` with the supplied name is performed and the result of the lookup is merged with the item and emitted.
    `StreamStage<T>` | `merge(StreamStage<? extends T> other)` Attaches a stage that emits all the items from this stage as well as all the items from the supplied stage.
    `default StreamStage<T>` | `peek()` Adds a peeking layer to this compute stage which logs its output.
    `default StreamStage<T>` | `peek(FunctionEx<? super T,? extends CharSequence> toStringFn)` Adds a peeking layer to this compute stage which logs its output.
    `StreamStage<T>` | `peek(PredicateEx<? super T> shouldLogFn, FunctionEx<? super T,? extends CharSequence> toStringFn)` Attaches a peeking stage which logs this stage's output and passes it through without transformation.
    `default <A,R> StreamStage<R>` | `rollingAggregate(AggregateOperation1<? super T,A,? extends R> aggrOp)` Attaches a rolling aggregation stage.
    `StreamStage<T>` | `setLocalParallelism(int localParallelism)` Sets the preferred local parallelism (number of processors per Jet cluster member) this stage will configure its DAG vertices with.
    `StreamStage<T>` | `setName(String name)` Overrides the default name of the stage with the name you choose and returns the stage.
    `StageWithWindow<T>` | `window(WindowDefinition wDef)` Adds the given window definition to this stage, as the first step in the construction of a pipeline stage that performs windowed aggregation.

    - ### Methods inherited from interface com.hazelcast.jet.pipeline.GeneralStage

      `addTimestamps, drainTo`

    - ### Methods inherited from interface com.hazelcast.jet.pipeline.Stage

      `getPipeline, name`

- - ### Method Detail

    - #### window

          @Nonnull
          StageWithWindow<T> window(WindowDefinition wDef)

      Adds the given window definition to this stage, as the first step in the construction of a pipeline stage that performs windowed aggregation.
      See Also:
        `factory methods in WindowDefiniton`

    - #### merge

          @Nonnull
          StreamStage<T> merge(@Nonnull
                                        StreamStage<? extends T> other)

      Attaches a stage that emits all the items from this stage as well as all the items from the supplied stage. The other stage's type parameter must be assignment-compatible with this stage's type parameter.
      Parameters:
        `other` - the other stage whose data to merge into this one
      Returns:
        the newly attached stage

    - #### groupingKey

          @Nonnull
          <K> StreamStageWithKey<T,K> groupingKey(@Nonnull
                                                           FunctionEx<? super T,? extends K> keyFn)

      Description copied from interface: `GeneralStage`
      Specifies the function that will extract a key from the items in the associated pipeline stage. This enables the operations that need the key, such as grouped aggregation.

      Sample usage:

           users.groupingKey(User::getId)

      Note: make sure the extracted key is not-null, it would fail the job otherwise. Also make sure that it implements `equals()` and `hashCode()`.

      Specified by:
        `groupingKey` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - type of the key
      Parameters:
        `keyFn` - function that extracts the grouping key
      Returns:
        the newly attached stage

    - #### map

          @Nonnull
          <R> StreamStage<R> map(@Nonnull
                                          FunctionEx<? super T,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage which applies the given function to each input item independently and emits the function's result as the output item. If the result is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      This sample takes a stream of names and outputs the names in lowercase:

           stage.map(name -> name.toLowerCase())

      Specified by:
        `map` in interface `GeneralStage<T>`
      Type Parameters:
        `R` - the result type of the mapping function
      Parameters:
        `mapFn` - a stateless mapping function
      Returns:
        the newly attached stage

    - #### filter

          @Nonnull
          StreamStage<T> filter(@Nonnull
                                         PredicateEx<T> filterFn)

      Description copied from interface: `GeneralStage`
      Attaches a filtering stage which applies the provided predicate function to each input item to decide whether to pass the item to the output or to discard it. Returns the newly attached stage.

      This sample removes empty strings from the stream:

           stage.filter(name -> !name.isEmpty())

      Specified by:
        `filter` in interface `GeneralStage<T>`
      Parameters:
        `filterFn` - a stateless filter predicate function
      Returns:
        the newly attached stage

    - #### flatMap

          @Nonnull
          <R> StreamStage<R> flatMap(@Nonnull
                                              FunctionEx<? super T,? extends Traverser<? extends R>> flatMapFn)

      Description copied from interface: `GeneralStage`
      Attaches a flat-mapping stage which applies the supplied function to each input item independently and emits all the items from the `Traverser` it returns. The traverser must be null-terminated.

      This sample takes a stream of sentences and outputs a stream of individual words in them:

           stage.map(sentence -> traverseArray(sentence.split("\\W+")))

      Specified by:
        `flatMap` in interface `GeneralStage<T>`
      Type Parameters:
        `R` - the type of items in the result's traversers
      Parameters:
        `flatMapFn` - a stateless flatmapping function, whose result type is Jet's `Traverser`. It must not return null traverser, but can return an empty traverser.
      Returns:
        the newly attached stage

    - #### mapStateful

          @Nonnull
          <S,R> StreamStage<R> mapStateful(@Nonnull
                                                    SupplierEx<? extends S> createFn,
                                                    @Nonnull
                                                    BiFunctionEx<? super S,? super T,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a stage that performs a stateful mapping operation. `createFn` returns the object that holds the state. Jet passes this object along with each input item to `mapFn`, which can update the object's state. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      This sample takes a stream of `long` numbers representing request latencies, computes the cumulative latency of all requests so far, and starts emitting alarm messages when the cumulative latency crosses a "bad behavior" threshold:

           StreamStage<Long> latencyAlarms = latencies.mapStateful(
                   LongAccumulator::new,
                   (sum, latency) -> {
                       sum.add(latency);
                       long cumulativeLatency = sum.get();
                       return (cumulativeLatency <= LATENCY_THRESHOLD)
                               ? null
                               : cumulativeLatency;
                   }
           );

      This code has the same result as `latencies.rollingAggregate(summing())`.
      Specified by:
        `mapStateful` in interface `GeneralStage<T>`
      Type Parameters:
        `S` - type of the state object
        `R` - type of the result
      Parameters:
        `createFn` - function that returns the state object
        `mapFn` - function that receives the state object and the input item and outputs the result item. It may modify the state object.

    - #### filterStateful

          @Nonnull
          <S> StreamStage<T> filterStateful(@Nonnull
                                                     SupplierEx<? extends S> createFn,
                                                     @Nonnull
                                                     BiPredicateEx<? super S,? super T> filterFn)

      Description copied from interface: `GeneralStage`
      Attaches a stage that performs a stateful filtering operation. `createFn` returns the object that holds the state. Jet passes this object along with each input item to `filterFn`, which can update the object's state. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      This sample decimates the input (throws out every 10th item):

           GeneralStage<String> decimated = input.filterStateful(
                   LongAccumulator::new,
                   (counter, item) -> {
                       counter.add(1);
                       return counter.get() % 10 != 0;
                   }
           );

      Specified by:
        `filterStateful` in interface `GeneralStage<T>`
      Type Parameters:
        `S` - type of the state object
      Parameters:
        `createFn` - function that returns the state object
        `filterFn` - function that receives the state object and the input item and produces the boolean result. It may modify the state object.

    - #### flatMapStateful

          @Nonnull
          <S,R> StreamStage<R> flatMapStateful(@Nonnull
                                                        SupplierEx<? extends S> createFn,
                                                        @Nonnull
                                                        BiFunctionEx<? super S,? super T,? extends Traverser<R>> flatMapFn)

      Description copied from interface: `GeneralStage`
      Attaches a stage that performs a stateful flat-mapping operation. `createFn` returns the object that holds the state. Jet passes this object along with each input item to `flatMapFn`, which can update the object's state. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      This sample inserts a punctuation mark (a special string) after every 10th input string:

           GeneralStage<String> punctuated = input.flatMapStateful(
                   LongAccumulator::new,
                   (counter, item) -> {
                       counter.add(1);
                       return counter.get() % 10 == 0
                               ? Traversers.traverseItems("punctuation", item)
                               : Traversers.singleton(item);
                   }
           );

      Specified by:
        `flatMapStateful` in interface `GeneralStage<T>`
      Type Parameters:
        `S` - type of the state object
        `R` - type of the result
      Parameters:
        `createFn` - function that returns the state object
        `flatMapFn` - function that receives the state object and the input item and outputs the result items. It may modify the state object. It must not return null traverser, but can return an empty traverser.

    - #### rollingAggregate

          @Nonnull
          default <A,R> StreamStage<R> rollingAggregate(@Nonnull
                                                                 AggregateOperation1<? super T,A,? extends R> aggrOp)

      Description copied from interface: `GeneralStage`
      Attaches a rolling aggregation stage. This is a special case of stateful mapping that uses an `AggregateOperation`. It passes each input item to the accumulator and outputs the current result of aggregation (as returned by the `export` primitive).

      Sample usage:

           stage.rollingAggregate(AggregateOperations.summing())

      For example, if your input is `{2, 7, 8, -5}`, the output will be `{2, 9, 17, 12}`.

      This stage is fault-tolerant and saves its state to the snapshot.

      NOTE: since the output for each item depends on all the previous items, this operation cannot be parallelized. Jet will perform it on a single member, single-threaded. Jet also supports `keyed rolling aggregation` which it can parallelize by partitioning.

      Specified by:
        `rollingAggregate` in interface `GeneralStage<T>`
        `R` - result type of the aggregate operation
      Parameters:
        `aggrOp` - the aggregate operation to do the aggregation
      Returns:
        the newly attached stage

    - #### mapUsingContext

          @Nonnull
          <C,R> StreamStage<R> mapUsingContext(@Nonnull
                                                        ContextFactory<C> contextFactory,
                                                        @Nonnull
                                                        BiFunctionEx<? super C,? super T,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage which applies the supplied function to each input item independently and emits the function's result as the output item. The mapping function receives another parameter, the context object, which Jet will create using the supplied `contextFactory`.

      If the mapping result is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           stage.mapUsingContext(
               ContextFactory.withCreateFn(jet -> new ItemDetailRegistry(jet)),
               (reg, item) -> item.setDetail(reg.fetchDetail(item))
           )

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `mapUsingContext` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
        `R` - the result type of the mapping function
      Parameters:
        `contextFactory` - the context factory
        `mapFn` - a stateless mapping function
      Returns:
        the newly attached stage

    - #### mapUsingContextAsync

          @Nonnull
          <C,R> StreamStage<R> mapUsingContextAsync(@Nonnull
                                                             ContextFactory<C> contextFactory,
                                                             @Nonnull
                                                             BiFunctionEx<? super C,? super T,? extends CompletableFuture<R>> mapAsyncFn)

      Description copied from interface: `GeneralStage`
      Asynchronous version of `GeneralStage.mapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiFunctionEx<? super C, ? super T, ? extends R>)`: the `mapAsyncFn` returns a `CompletableFuture<R>` instead of just `R`.

      The function can return a null future or the future can return a null result: in both cases it will act just like a filter.

      The latency of the async call will add to the total latency of the output.

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           stage.mapUsingContextAsync(
               ContextFactory.withCreateFn(jet -> new ItemDetailRegistry(jet)),
               (reg, item) -> reg.fetchDetailAsync(item)
                                 .thenApply(detail -> item.setDetail(detail)
           )

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `mapUsingContextAsync` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
        `R` - the future's result type of the mapping function
      Parameters:
        `contextFactory` - the context factory
        `mapAsyncFn` - a stateless mapping function. Can map to null (return a null future)
      Returns:
        the newly attached stage

    - #### filterUsingContext

          @Nonnull
          <C> StreamStage<T> filterUsingContext(@Nonnull
                                                         ContextFactory<C> contextFactory,
                                                         @Nonnull
                                                         BiPredicateEx<? super C,? super T> filterFn)

      Description copied from interface: `GeneralStage`
      Attaches a filtering stage which applies the provided predicate function to each input item to decide whether to pass the item to the output or to discard it. The predicate function receives another parameter, the context object, which Jet will create using the supplied `contextFactory`.

      This sample takes a stream of photos, uses an image classifier to reason about their contents, and keeps only photos of cats:

           photos.filterUsingContext(
               ContextFactory.withCreateFn(jet -> new ImageClassifier(jet)),
               (classifier, photo) -> classifier.classify(photo).equals("cat")
           )

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `filterUsingContext` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
      Parameters:
        `contextFactory` - the context factory
        `filterFn` - a stateless filter predicate function
      Returns:
        the newly attached stage

    - #### filterUsingContextAsync

          @Nonnull
          <C> StreamStage<T> filterUsingContextAsync(@Nonnull
                                                              ContextFactory<C> contextFactory,
                                                              @Nonnull
                                                              BiFunctionEx<? super C,? super T,? extends CompletableFuture<Boolean>> filterAsyncFn)

      Description copied from interface: `GeneralStage`
      Asynchronous version of `GeneralStage.filterUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiPredicateEx<? super C, ? super T>)`: the `filterAsyncFn` returns a `CompletableFuture<Boolean>` instead of just a `boolean`.

      The function must not return a null future.

      The latency of the async call will add to the total latency of the output.

      This sample takes a stream of photos, uses an image classifier to reason about their contents, and keeps only photos of cats:

           photos.filterUsingContextAsync(
               ContextFactory.withCreateFn(jet -> new ImageClassifier(jet)),
               (classifier, photo) -> reg.classifyAsync(photo)
                                         .thenApply(it -> it.equals("cat"))
           )

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `filterUsingContextAsync` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
      Parameters:
        `contextFactory` - the context factory
        `filterAsyncFn` - a stateless filtering function
      Returns:
        the newly attached stage

    - #### flatMapUsingContext

          @Nonnull
          <C,R> StreamStage<R> flatMapUsingContext(@Nonnull
                                                            ContextFactory<C> contextFactory,
                                                            @Nonnull
                                                            BiFunctionEx<? super C,? super T,? extends Traverser<R>> flatMapFn)

      Description copied from interface: `GeneralStage`
      Attaches a flat-mapping stage which applies the supplied function to each input item independently and emits all items from the `Traverser` it returns as the output items. The traverser must be null-terminated. The mapping function receives another parameter, the context object, which Jet will create using the supplied `contextFactory`.

      This sample takes a stream of products and outputs an "exploded" stream of all the parts that go into making them:

           StreamStage<Part> parts = products.flatMapUsingContext(
               ContextFactory.withCreateFn(jet -> new PartRegistryCtx()),
               (registry, product) -> Traversers.traverseIterable(
                                          registry.fetchParts(product))
           );

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `flatMapUsingContext` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
        `R` - the type of items in the result's traversers
      Parameters:
        `contextFactory` - the context factory
        `flatMapFn` - a stateless flatmapping function, whose result type is Jet's `Traverser`. It must not return null traverser, but can return an empty traverser.
      Returns:
        the newly attached stage

    - #### flatMapUsingContextAsync

          @Nonnull
          <C,R> StreamStage<R> flatMapUsingContextAsync(@Nonnull
                                                                 ContextFactory<C> contextFactory,
                                                                 @Nonnull
                                                                 BiFunctionEx<? super C,? super T,? extends CompletableFuture<Traverser<R>>> flatMapAsyncFn)

      Description copied from interface: `GeneralStage`
      Asynchronous version of `GeneralStage.flatMapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.BiFunctionEx<? super C, ? super T, ? extends com.hazelcast.jet.Traverser<R>>)`: the `flatMapAsyncFn` returns a `CompletableFuture<Traverser<R>>` instead of just `Traverser<R>`.

      The function can return a null future or the future can return a null traverser: in both cases it will act just like a filter.

      The latency of the async call will add to the total latency of the output.

      This sample takes a stream of products and outputs an "exploded" stream of all the parts that go into making them:

           StreamStage<Part> parts = products.flatMapUsingContextAsync(
               ContextFactory.withCreateFn(jet -> new PartRegistryCtx()),
               (registry, product) -> registry
                    .fetchPartsAsync(product)
                    .thenApply(parts -> Traversers.traverseIterable(parts))
           );

      ### Interaction with fault-tolerant unbounded jobs

      If you use this stage in a fault-tolerant unbounded job, keep in mind that any state the context object maintains doesn't participate in Jet's fault tolerance protocol. If the state is local, it will be lost after a job restart; if it is saved to some durable storage, the state of that storage won't be rewound to the last checkpoint, so you'll perform duplicate updates.
      Specified by:
        `flatMapUsingContextAsync` in interface `GeneralStage<T>`
      Type Parameters:
        `C` - type of context object
        `R` - the type of the returned stage
      Parameters:
        `contextFactory` - the context factory
        `flatMapAsyncFn` - a stateless flatmapping function. Can map to null (return a null future). The future must not return a null traverser, but can return an empty traverser.
      Returns:
        the newly attached stage

    - #### mapUsingReplicatedMap

          @Nonnull
          default <K,V,R> StreamStage<R> mapUsingReplicatedMap(@Nonnull
                                                                        String mapName,
                                                                        @Nonnull
                                                                        FunctionEx<? super T,? extends K> lookupKeyFn,
                                                                        @Nonnull
                                                                        BiFunctionEx<? super T,? super V,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage where for each item a lookup in the `ReplicatedMap` with the supplied name is performed and the result of the lookup is merged with the item and emitted.

      If the result of the mapping is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      The mapping logic is equivalent to:

           K key = lookupKeyFn.apply(item);
           V value = replicatedMap.get(key);
           return mapFn.apply(item, value);

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           items.mapUsingReplicatedMap(
               "enriching-map",
               item -> item.getDetailId(),
               (Item item, ItemDetail detail) -> item.setDetail(detail)
           )

      Specified by:
        `mapUsingReplicatedMap` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - type of the key in the `ReplicatedMap`
        `V` - type of the value in the `ReplicatedMap`
        `R` - type of the output item
      Parameters:
        `mapName` - name of the `ReplicatedMap`
        `lookupKeyFn` - a function which returns the key to look up in the map. Must not return null
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### mapUsingReplicatedMap

          @Nonnull
          default <K,V,R> StreamStage<R> mapUsingReplicatedMap(@Nonnull
                                                                        ReplicatedMap<K,V> replicatedMap,
                                                                        @Nonnull
                                                                        FunctionEx<? super T,? extends K> lookupKeyFn,
                                                                        @Nonnull
                                                                        BiFunctionEx<? super T,? super V,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage where for each item a lookup in the supplied `ReplicatedMap` is performed and the result of the lookup is merged with the item and emitted.

      If the result of the mapping is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      The mapping logic is equivalent to:

           K key = lookupKeyFn.apply(item);
           V value = replicatedMap.get(key);
           return mapFn.apply(item, value);

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           items.mapUsingReplicatedMap(
               enrichingMap,
               item -> item.getDetailId(),
               (item, detail) -> item.setDetail(detail)
           )

      Specified by:
        `mapUsingReplicatedMap` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - type of the key in the `ReplicatedMap`
        `V` - type of the value in the `ReplicatedMap`
        `R` - type of the output item
      Parameters:
        `replicatedMap` - the `ReplicatedMap` to lookup from
        `lookupKeyFn` - a function which returns the key to look up in the map. Must not return null
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### mapUsingIMap

          @Nonnull
          default <K,V,R> StreamStage<R> mapUsingIMap(@Nonnull
                                                               String mapName,
                                                               @Nonnull
                                                               FunctionEx<? super T,? extends K> lookupKeyFn,
                                                               @Nonnull
                                                               BiFunctionEx<? super T,? super V,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage where for each item a lookup in the `IMap` with the supplied name is performed and the result of the lookup is merged with the item and emitted.

      If the result of the mapping is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      The mapping logic is equivalent to:

           K key = lookupKeyFn.apply(item);
           V value = map.get(key);
           return mapFn.apply(item, value);

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           items.mapUsingIMap(
               "enriching-map",
               item -> item.getDetailId(),
               (Item item, ItemDetail detail) -> item.setDetail(detail)
           )

      See also `GeneralStageWithKey.mapUsingIMap(java.lang.String, com.hazelcast.jet.function.BiFunctionEx<? super T, ? super V, ? extends R>)` for a partitioned version of this operation.
      Specified by:
        `mapUsingIMap` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - type of the key in the `IMap`
        `V` - type of the value in the `IMap`
        `R` - type of the output item
      Parameters:
        `mapName` - name of the `IMap`
        `lookupKeyFn` - a function which returns the key to look up in the map. Must not return null
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### mapUsingIMap

          @Nonnull
          default <K,V,R> StreamStage<R> mapUsingIMap(@Nonnull
                                                               IMap<K,V> iMap,
                                                               @Nonnull
                                                               FunctionEx<? super T,? extends K> lookupKeyFn,
                                                               @Nonnull
                                                               BiFunctionEx<? super T,? super V,? extends R> mapFn)

      Description copied from interface: `GeneralStage`
      Attaches a mapping stage where for each item a lookup in the supplied `IMap` is performed and the result of the lookup is merged with the item and emitted.

      If the result of the mapping is `null`, it emits nothing. Therefore this stage can be used to implement filtering semantics as well.

      The mapping logic is equivalent to:

           K key = lookupKeyFn.apply(item);
           V value = map.get(key);
           return mapFn.apply(item, value);

      This sample takes a stream of stock items and sets the `detail` field on them by looking up from a registry:

           items.mapUsingIMap(
               enrichingMap,
               item -> item.getDetailId(),
               (item, detail) -> item.setDetail(detail)
           )

      See also `GeneralStageWithKey.mapUsingIMap(java.lang.String, com.hazelcast.jet.function.BiFunctionEx<? super T, ? super V, ? extends R>)` for a partitioned version of this operation.
      Specified by:
        `mapUsingIMap` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - type of the key in the `IMap`
        `V` - type of the value in the `IMap`
        `R` - type of the output item
      Parameters:
        `iMap` - the `IMap` to lookup from
        `lookupKeyFn` - a function which returns the key to look up in the map. Must not return null
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### hashJoin

          @Nonnull
          <K,T1_IN,T1,R> StreamStage<R> hashJoin(@Nonnull
                                                          BatchStage<T1_IN> stage1,
                                                          @Nonnull
                                                          JoinClause<K,? super T,? super T1_IN,? extends T1> joinClause1,
                                                          @Nonnull
                                                          BiFunctionEx<T,T1,R> mapToOutputFn)

      Description copied from interface: `GeneralStage`
      Attaches to both this and the supplied stage a hash-joining stage and returns it. This stage plays the role of the primary stage in the hash-join. Please refer to the `package javadoc` for a detailed description of the hash-join transform.

      This sample joins a stream of users to a stream of countries and outputs a stream of users with the `country` field set:

           // Types of the input stages:
           BatchStage<User> users;
           BatchStage<Map.Entry<Long, Country>> idAndCountry;

           users.hashJoin(
               idAndCountry,
               JoinClause.joinMapEntries(User::getCountryId),
               (user, country) -> user.setCountry(country)
           )

      Specified by:
        `hashJoin` in interface `GeneralStage<T>`
      Type Parameters:
        `K` - the type of the join key
        `T1_IN` - the type of `stage1` items
        `T1` - the result type of projection on `stage1` items
        `R` - the resulting output type
      Parameters:
        `stage1` - the stage to hash-join with this one
        `joinClause1` - specifies how to join the two streams
        `mapToOutputFn` - function to map the joined items to the output value
      Returns:
        the newly attached stage

    - #### hashJoin2

          @Nonnull
          <K1,K2,T1_IN,T2_IN,T1,T2,R> StreamStage<R> hashJoin2(@Nonnull
                                                                        BatchStage<T1_IN> stage1,
                                                                        @Nonnull
                                                                        JoinClause<K1,? super T,? super T1_IN,? extends T1> joinClause1,
                                                                        @Nonnull
                                                                        BatchStage<T2_IN> stage2,
                                                                        @Nonnull
                                                                        JoinClause<K2,? super T,? super T2_IN,? extends T2> joinClause2,
                                                                        @Nonnull
                                                                        TriFunction<T,T1,T2,R> mapToOutputFn)

      Description copied from interface: `GeneralStage`
      Attaches to this and the two supplied stages a hash-joining stage and returns it. This stage plays the role of the primary stage in the hash-join. Please refer to the `package javadoc` for a detailed description of the hash-join transform.

      This sample joins a stream of users to streams of countries and companies, and outputs a stream of users with the `country` and `company` fields set:

           // Types of the input stages:
           BatchStage<User> users;
           BatchStage<Map.Entry<Long, Country>> idAndCountry;
           BatchStage<Map.Entry<Long, Company>> idAndCompany;

           users.hashJoin(
               idAndCountry, JoinClause.joinMapEntries(User::getCountryId),
               idAndCompany, JoinClause.joinMapEntries(User::getCompanyId),
               (user, country, company) -> user.setCountry(country).setCompany(company)
           )

      Specified by:
        `hashJoin2` in interface `GeneralStage<T>`
      Type Parameters:
        `K1` - the type of key for `stage1`
        `K2` - the type of key for `stage2`
        `T1_IN` - the type of `stage1` items
        `T2_IN` - the type of `stage2` items
        `T1` - the result type of projection of `stage1` items
        `T2` - the result type of projection of `stage2` items
        `R` - the resulting output type
      Parameters:
        `stage1` - the first stage to join
        `joinClause1` - specifies how to join with `stage1`
        `stage2` - the second stage to join
        `joinClause2` - specifies how to join with `stage2`
        `mapToOutputFn` - function to map the joined items to the output value
      Returns:
        the newly attached stage

    - #### hashJoinBuilder

          @Nonnull
          default StreamHashJoinBuilder<T> hashJoinBuilder()

      Description copied from interface: `GeneralStage`
      Returns a fluent API builder object to construct a hash join operation with any number of contributing stages. It is mainly intended for hash-joins with three or more enriching stages. For one or two stages prefer the direct `stage.hashJoinN(...)` calls because they offer more static type safety.

      This sample joins a stream of users to streams of countries and companies, and outputs a stream of users with the `country` and `company` fields set:

           // Types of the input stages:
           StreamStage<User> users;
           BatchStage<Map.Entry<Long, Country>> idAndCountry;
           BatchStage<Map.Entry<Long, Company>> idAndCompany;

           StreamHashJoinBuilder<User> builder = users.hashJoinBuilder();
           Tag<Country> tCountry = builder.add(idAndCountry,
                   JoinClause.joinMapEntries(User::getCountryId));
           Tag<Company> tCompany = builder.add(idAndCompany,
                   JoinClause.joinMapEntries(User::getCompanyId));
           StreamStage<User> joined = builder.build((user, itemsByTag) ->
                   user.setCountry(itemsByTag.get(tCountry)).setCompany(itemsByTag.get(tCompany)));

      Specified by:
        `hashJoinBuilder` in interface `GeneralStage<T>`
      Returns:
        the newly attached stage

    - #### peek

          @Nonnull
          default StreamStage<T> peek()

      Description copied from interface: `GeneralStage`
      Adds a peeking layer to this compute stage which logs its output. For each item the stage emits, it logs the result of its `toString()` method at the INFO level to the log category `com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>`. The stage logs each item on whichever cluster member it happens to receive it. Its primary purpose is for development use, when running Jet on a local machine.
      Specified by:
        `peek` in interface `GeneralStage<T>`
      Returns:
        the newly attached stage
      See Also:
        `GeneralStage.peek(PredicateEx, FunctionEx)`, `GeneralStage.peek(FunctionEx)`

    - #### peek

          @Nonnull
          StreamStage<T> peek(@Nonnull
                                       PredicateEx<? super T> shouldLogFn,
                                       @Nonnull
                                       FunctionEx<? super T,? extends CharSequence> toStringFn)

      Description copied from interface: `GeneralStage`
      Attaches a peeking stage which logs this stage's output and passes it through without transformation. For each item the stage emits, it:
      1. uses the `shouldLogFn` predicate to see whether to log the item
      2. if yes, uses then uses `toStringFn` to get the item's string representation
      3. logs the string at the INFO level to the log category `com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>`
      The stage logs each item on whichever cluster member it happens to receive it. Its primary purpose is for development use, when running Jet on a local machine.

      Sample usage:

           users.peek(
               user -> user.getName().size() > 100,
               User::getName
           )

      Specified by:
        `peek` in interface `GeneralStage<T>`
      Parameters:
        `shouldLogFn` - a function to filter the logged items. You can use `alwaysTrue()` as a pass-through filter when you don't need any filtering.
        `toStringFn` - a function that returns a string representation of the item
      Returns:
        the newly attached stage
      See Also:
        `GeneralStage.peek(FunctionEx)`, `GeneralStage.peek()`

    - #### peek

          @Nonnull
          default StreamStage<T> peek(@Nonnull
                                               FunctionEx<? super T,? extends CharSequence> toStringFn)

      Description copied from interface: `GeneralStage`
      Adds a peeking layer to this compute stage which logs its output. For each item the stage emits, it:
      1. uses `toStringFn` to get a string representation of the item
      2. logs the string at the INFO level to the log category `com.hazelcast.jet.impl.processor.PeekWrappedP.<vertexName>#<processorIndex>`
      The stage logs each item on whichever cluster member it happens to receive it. Its primary purpose is for development use, when running Jet on a local machine.

      Sample usage:

           users.peek(User::getName)

      Specified by:
        `peek` in interface `GeneralStage<T>`
      Parameters:
        `toStringFn` - a function that returns a string representation of the item
      Returns:
        the newly attached stage
      See Also:
        `GeneralStage.peek(PredicateEx, FunctionEx)`, `GeneralStage.peek()`

    - #### customTransform

          @Nonnull
          default <R> StreamStage<R> customTransform(@Nonnull
                                                              String stageName,
                                                              @Nonnull
                                                              SupplierEx<Processor> procSupplier)

      Description copied from interface: `GeneralStage`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.

      Note that the type parameter of the returned stage is inferred from the call site and not propagated from the processor that will produce the result, so there is no actual type safety provided.

      Specified by:
        `customTransform` in interface `GeneralStage<T>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors
      Returns:
        the newly attached stage

    - #### customTransform

          @Nonnull
          default <R> StreamStage<R> customTransform(@Nonnull
                                                              String stageName,
                                                              @Nonnull
                                                              ProcessorSupplier procSupplier)

      Description copied from interface: `GeneralStage`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.

      Note that the type parameter of the returned stage is inferred from the call site and not propagated from the processor that will produce the result, so there is no actual type safety provided.

      Specified by:
        `customTransform` in interface `GeneralStage<T>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors
      Returns:
        the newly attached stage

    - #### customTransform

          @Nonnull
          <R> StreamStage<R> customTransform(@Nonnull
                                                      String stageName,
                                                      @Nonnull
                                                      ProcessorMetaSupplier procSupplier)

      Description copied from interface: `GeneralStage`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s.

      Note that the type parameter of the returned stage is inferred from the call site and not propagated from the processor that will produce the result, so there is no actual type safety provided.

      Specified by:
        `customTransform` in interface `GeneralStage<T>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors
      Returns:
        the newly attached stage

    - #### apply

          @Nonnull
          default <R> StreamStage<R> apply(@Nonnull
                                                    FunctionEx<? super StreamStage<T>,? extends StreamStage<R>> transformFn)

      Transforms `this` stage using the provided `transformFn` and returns the transformed stage. It allows you to extract common pipeline transformations into a method and then call that method without interrupting the chained pipeline expression.

      For example, say you have this code:

           StreamStage<String> input = pipeline.drawFrom(textSource);
           StreamStage<String> cleanedUp = input
                   .map(String::toLowerCase)
                   .filter(s -> s.startsWith("success"));

      You can capture the `map` and `filter` steps into a common "cleanup" transformation:

           StreamStage<String> cleanUp(StreamStage<String> input) {
                return input.map(String::toLowerCase)
                            .filter(s -> s.startsWith("success"));
           }

      Now you can insert this transformation as just another step in your pipeline:

           StreamStage<String> tokens = pipeline
               .drawFrom(textSource)
               .apply(this::cleanUp)
               .flatMap(line -> traverseArray(line.split("\\W+")));

      Type Parameters:
        `R` - type of the returned stage
      Parameters:
        `transformFn` - function to transform this stage into another stage
      Since:
        3.1

    - #### setLocalParallelism

          @Nonnull
          StreamStage<T> setLocalParallelism(int localParallelism)

      Description copied from interface: `Stage`
      Sets the preferred local parallelism (number of processors per Jet cluster member) this stage will configure its DAG vertices with. Jet always uses the same number of processors on each member, so the total parallelism automatically increases if another member joins the cluster.

      While most stages are backed by 1 vertex, there are exceptions. If a stage uses two vertices, each of them will have the given local parallelism, so in total there will be twice as many processors per member.

      The default value is -1 and it signals to Jet to figure out a default value. Jet will determine the vertex's local parallelism during job initialization from the global default and the processor meta-supplier's preferred value.

      Specified by:
        `setLocalParallelism` in interface `Stage`
      Returns:
        this stage

    - #### setName

          @Nonnull
          StreamStage<T> setName(@Nonnull
                                          String name)

      Description copied from interface: `Stage`
      Overrides the default name of the stage with the name you choose and returns the stage. This can be useful for debugging purposes, to better distinguish pipeline stages in the diagnostic output.
      Specified by:
        `setName` in interface `GeneralStage<T>`
      Specified by:
        `setName` in interface `Stage`
      Parameters:
        `name` - the stage name
      Returns:
        this stage

Copyright © 2020. All rights reserved.