com.hazelcast.jet.pipeline

## Interface StreamStageWithKey<T,K>

- Type Parameters:
    `T` - type of the stream items
    `K` - type of the key
  All Superinterfaces:
    GeneralStageWithKey<T,K>

      public interface StreamStageWithKey<T,K>
      extends GeneralStageWithKey<T,K>

  An intermediate step while constructing a pieline transform that involves a grouping key, such as windowed group-and-aggregate. Some transforms use the grouping key only to partition the stream (such as `mapUsingIMap(java.lang.String, com.hazelcast.jet.function.BiFunctionEx<? super T, ? super V, ? extends R>)`).
  Since:
    3.0

- - ### Method Summary

    All Methods Instance Methods Abstract Methods Default Methods
    Modifier and Type | Method and Description
    --- | ---
    `<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.
    `<S> StreamStage<T>` | `filterStateful(long ttl, SupplierEx<? extends S> createFn, BiPredicateEx<? super S,? super T> filterFn)` Attaches a stage that performs a stateful filtering operation.
    `default <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, TriPredicate<? super C,? super K,? 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, TriFunction<? super C,? super K,? super T,CompletableFuture<Boolean>> filterAsyncFn)` Asynchronous version of `GeneralStageWithKey.filterUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.TriPredicate<? super C, ? super K, ? super T>)`: the `filterAsyncFn` returns a `CompletableFuture<Boolean>` instead of just a `boolean`.
    `<S,R> StreamStage<R>` | `flatMapStateful(long ttl, SupplierEx<? extends S> createFn, TriFunction<? super S,? super K,? super T,? extends Traverser<R>> flatMapFn, TriFunction<? super S,? super K,? super Long,? extends Traverser<R>> onEvictFn)` Attaches a stage that performs a stateful flat-mapping operation.
    `<S,R> StreamStage<R>` | `flatMapStateful(SupplierEx<? extends S> createFn, TriFunction<? super S,? super K,? super T,? extends Traverser<R>> flatMapFn)` Attaches a stage that performs a stateful flat-mapping operation.
    `<C,R> StreamStage<R>` | `flatMapUsingContext(ContextFactory<C> contextFactory, TriFunction<? super C,? super K,? 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 as the output items.
    `<C,R> StreamStage<R>` | `flatMapUsingContextAsync(ContextFactory<C> contextFactory, TriFunction<? super C,? super K,? super T,CompletableFuture<Traverser<R>>> flatMapAsyncFn)` Asynchronous version of `GeneralStageWithKey.flatMapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.TriFunction<? super C, ? super K, ? super T, ? extends com.hazelcast.jet.Traverser<? extends R>>)`: the `flatMapAsyncFn` returns a `CompletableFuture<Traverser<R>>` instead of just `Traverser<R>`.
    `<S,R> StreamStage<R>` | `mapStateful(long ttl, SupplierEx<? extends S> createFn, TriFunction<? super S,? super K,? super T,? extends R> mapFn, TriFunction<? super S,? super K,? super Long,? extends R> onEvictFn)` Attaches a stage that performs a stateful mapping operation.
    `<S,R> StreamStage<R>` | `mapStateful(SupplierEx<? extends S> createFn, TriFunction<? super S,? super K,? super T,? extends R> mapFn)` Attaches a stage that performs a stateful mapping operation.
    `<C,R> StreamStage<R>` | `mapUsingContext(ContextFactory<C> contextFactory, TriFunction<? super C,? super K,? 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.
    `<C,R> StreamStage<R>` | `mapUsingContextAsync(ContextFactory<C> contextFactory, TriFunction<? super C,? super K,? super T,CompletableFuture<R>> mapAsyncFn)` Asynchronous version of `GeneralStageWithKey.mapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.TriFunction<? super C, ? super K, ? super T, ? extends R>)`: the `mapAsyncFn` returns a `CompletableFuture<R>` instead of just `R`.
    `default <V,R> StreamStage<R>` | `mapUsingIMap(IMap<K,V> iMap, BiFunctionEx<? super T,? super V,? extends R> mapFn)` Attaches a mapping stage where for each item a lookup in the supplied `IMap` using the grouping key is performed and the result of the lookup is merged with the item and emitted.
    `default <V,R> StreamStage<R>` | `mapUsingIMap(String mapName, 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 using the grouping key is performed and the result of the lookup is merged with the item and emitted.
    `default <A,R> StreamStage<Map.Entry<K,R>>` | `rollingAggregate(AggregateOperation1<? super T,A,? extends R> aggrOp)` Attaches a rolling aggregation stage.
    `default <A,R> StreamStage<Map.Entry<K,R>>` | `rollingAggregate(long ttl, AggregateOperation1<? super T,A,? extends R> aggrOp)` Attaches a rolling aggregation stage.
    `StageWithKeyAndWindow<T,K>` | `window(WindowDefinition wDef)` Adds the definition of the window to use in the group-and-aggregate pipeline stage being constructed.

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

      `keyFn`

- - ### Method Detail

    - #### window

          @Nonnull
          StageWithKeyAndWindow<T,K> window(@Nonnull
                                                     WindowDefinition wDef)

      Adds the definition of the window to use in the group-and-aggregate pipeline stage being constructed.

    - #### mapStateful

          @Nonnull
          <S,R> StreamStage<R> mapStateful(long ttl,
                                                    @Nonnull
                                                    SupplierEx<? extends S> createFn,
                                                    @Nonnull
                                                    TriFunction<? super S,? super K,? super T,? extends R> mapFn,
                                                    @Nonnull
                                                    TriFunction<? super S,? super K,? super Long,? extends R> onEvictFn)

      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. For each grouping key there's a separate state object. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      If the given `ttl` is greater than zero, Jet will consider the state object stale if its time-to-live has expired. The state object for a given key has a timestamp attached to it: the top timestamp of any event with that key seen so far. Whenever the watermark advances, Jet discards all state objects with a timestamp less than `wm - ttl`. Just before discarding the state object, Jet calls `onEvictFn` on it. The function can return an output item that will be emitted, or `null` if it doesn't need to emit an item. If TTL is used, Jet also drops late events; otherwise, all events are processed.

      This sample takes a stream of pairs `(serverId, latency)` representing the latencies of serving individual requests and keeps track, separately for each server, of the total latency accumulated over individual sessions — bursts of server activity separated by quiet periods of one minute or more. For each input item it outputs the accumulated latency so far and when a session ends, it outputs a special entry that reports the total latency for that session.

           StreamStage<Entry<String, Long>> latencies = null;
           StreamStage<Entry<String, Long>> cumulativeLatencies = latencies
                   .groupingKey(Entry::getKey)
                   .mapStateful(
                           MINUTES.toMillis(1),
                           LongAccumulator::new,
                           (sum, key, entry) -> {
                               sum.add(entry.getValue());
                               return entry(key, sum.get());
                           },
                           (sum, key, time) -> entry(String.format(
                                   "%s:totalForSession:%d", key, time), sum.get())
                   );

      Type Parameters:
        `S` - type of the state object
        `R` - type of the result
      Parameters:
        `ttl` - time-to-live for each state object, disabled if zero or less
        `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.
        `onEvictFn` - function that Jet calls when evicting a state object

    - #### mapStateful

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

      Description copied from interface: `GeneralStageWithKey`
      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 and its key to `mapFn`, which can update the object's state. For each grouping key there's a separate state object. 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 pairs `(serverId, latency)` representing the latencies of serving individual requests and outputs the cumulative latency of all handled requests so far, for each server separately:

           GeneralStage<Entry<String, Long>> latencies;
           GeneralStage<Entry<String, Long>> cumulativeLatencies = latencies
                   .groupingKey(Entry::getKey)
                   .mapStateful(
                           LongAccumulator::new,
                           (sum, key, entry) -> {
                               sum.add(entry.getValue());
                               return entry(key, sum.get());
                           }
                   );

      This code has the same result as `latencies.groupingKey(Entry::getKey).rollingAggregate(summing())`.
      Specified by:
        `mapStateful` in interface `GeneralStageWithKey<T,K>`
      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(long ttl,
                                                     @Nonnull
                                                     SupplierEx<? extends S> createFn,
                                                     @Nonnull
                                                     BiPredicateEx<? super S,? super T> filterFn)

      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. For each grouping key there's a separate state object. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      If the given `ttl` is greater than zero, Jet will consider the state object stale if its time-to-live has expired. The state object has a timestamp attached to it: the top timestamp of any event with the same key seen so far. Upon seeing another event, Jet compares the state timestamp with the current watermark. If it is less than `wm - ttl`, it discards the state object and creates a new one before processing the event. If TTL is used, Jet also drops late events; otherwise, all events are processed.

      This sample receives a stream of pairs `(serverId, requestLatency)` that represent the latencies of individual requests served by a cluster of servers. It emits the record-breaking (worst so far) latencies for each server independently and resets the score after one minute of inactivity on a given server.

           StreamStage<Entry<String, Long>> latencies;
           StreamStage<Entry<String, Long>> topLatencies = latencies
                   .groupingKey(Entry::getKey)
                   .filterStateful(
                           MINUTES.toMillis(1),
                           LongAccumulator::new,
                           (topLatencyState, entry) -> {
                               long currLatency = entry.getValue();
                               long topLatency = topLatencyState.get();
                               topLatencyState.set(Math.max(currLatency, topLatency));
                               return currLatency > topLatency;
                           }
                   );

      Type Parameters:
        `S` - type of the state object
      Parameters:
        `ttl` - time-to-live for each state object, disabled if zero or less
        `createFn` - function that returns the state object
        `filterFn` - predicate that receives the state object and the input item and outputs a boolean value. It may modify the state object.

    - #### filterStateful

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

      Description copied from interface: `GeneralStageWithKey`
      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 and its key to `filterFn`, which can update the object's state. For each grouping key there's a separate state object. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      This sample groups a stream of strings by length and decimates each group (throws out every 10th string of each length):

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

      Specified by:
        `filterStateful` in interface `GeneralStageWithKey<T,K>`
      Type Parameters:
        `S` - type of the state object
      Parameters:
        `createFn` - function that returns the state object
        `filterFn` - predicate that receives the state object and the input item and outputs a boolean value. It may modify the state object.

    - #### flatMapStateful

          @Nonnull
          <S,R> StreamStage<R> flatMapStateful(long ttl,
                                                        @Nonnull
                                                        SupplierEx<? extends S> createFn,
                                                        @Nonnull
                                                        TriFunction<? super S,? super K,? super T,? extends Traverser<R>> flatMapFn,
                                                        @Nonnull
                                                        TriFunction<? super S,? super K,? super Long,? extends Traverser<R>> onEvictFn)

      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. For each grouping key there's a separate state object. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      If the given `ttl` is greater than zero, Jet will consider the state object stale if its time-to-live has expired. The state object for a given key has a timestamp attached to it: the top timestamp of any event with that key seen so far. Whenever the watermark advances, Jet discards all state objects with a timestamp less than `wm - ttl`. Just before discarding the state object, Jet calls `onEvictFn` on it. The function returns a traverser over the items it wants to emit, or it can return an empty traverser. If TTL is used, Jet also drops late events; otherwise, all events are processed.

      This sample groups a stream of strings by length and inserts punctuation (a special string) after every 10th string in each group, or after one minute elapses without further input for a given key:

           StreamStage<String> punctuated = input
                   .groupingKey(String::length)
                   .flatMapStateful(
                           MINUTES.toMillis(1),
                           LongAccumulator::new,
                           (counter, key, item) -> {
                               counter.add(1);
                               return counter.get() % 10 == 0
                                       ? Traversers.traverseItems("punctuation" + key, item)
                                       : Traversers.singleton(item);
                           },
                           (counter, key, wm) -> Traversers.singleton("punctuation" + key)
                   );

      Type Parameters:
        `S` - type of the state object
        `R` - type of the result
      Parameters:
        `ttl` - time-to-live for each state object, disabled if zero or less
        `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.
        `onEvictFn` - function that Jet calls when evicting a state object

    - #### flatMapStateful

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

      Description copied from interface: `GeneralStageWithKey`
      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 and its key to `flatMapFn`, which can update the object's state. For each grouping key there's a separate state object. The state object will be included in the state snapshot, so it survives job restarts. For this reason it must be serializable.

      This sample groups a stream of strings by length and inserts punctuation (a special string) after every 10th string in each group:

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

      Specified by:
        `flatMapStateful` in interface `GeneralStageWithKey<T,K>`
      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<Map.Entry<K,R>> rollingAggregate(@Nonnull
                                                                              AggregateOperation1<? super T,A,? extends R> aggrOp)

      Description copied from interface: `GeneralStageWithKey`
      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:

           StreamStage<Entry<Color, Long>> aggregated = items
                   .groupingKey(Item::getColor)
                   .rollingAggregate(AggregateOperations.counting());

      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.

      Specified by:
        `rollingAggregate` in interface `GeneralStageWithKey<T,K>`
        `R` - type of the aggregate operation result
      Parameters:
        `aggrOp` - the aggregate operation to perform
      Returns:
        the newly attached stage

    - #### rollingAggregate

          @Nonnull
          default <A,R> StreamStage<Map.Entry<K,R>> rollingAggregate(long ttl,
                                                                              @Nonnull
                                                                              AggregateOperation1<? super T,A,? extends R> aggrOp)

      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).

      This sample takes a stream of items and gives rolling counts of items of each color:

           StreamStage<Entry<Color, Long>> aggregated = items
                   .groupingKey(Item::getColor)
                   .rollingAggregate(AggregateOperations.counting());

      If the given `ttl` is greater than zero, Jet will consider the accumulator object stale if its time-to-live has expired. The accumulator object has a timestamp attached to it: the top timestamp of any event with the same key seen so far. Upon seeing another event, Jet compares the accumulator timestamp with the current watermark. If it is less than `wm - ttl`, it discards the accumulator object and creates a new one before processing the event.

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

      Type Parameters:
        `R` - type of the aggregate operation result
      Parameters:
        `aggrOp` - the aggregate operation to perform
      Returns:
        the newly attached stage

    - #### mapUsingIMap

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a mapping stage where for each item a lookup in the `IMap` with the supplied name using the grouping key 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:

           V value = map.get(groupingKey);
           return mapFn.apply(item, value);

      Sample usage:

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

      This stage is similar to `stageWithoutKey.mapUsingIMap()`, but here Jet knows the key and uses it to partition and distribute the input in order to achieve data locality. The value it fetches from the `IMap` is stored on the cluster member where the processing takes place. However, if the map doesn't use the default partitioning strategy, the data locality will be broken.
      Specified by:
        `mapUsingIMap` in interface `GeneralStageWithKey<T,K>`
      Type Parameters:
        `V` - type of the value in the `IMap`
        `R` - type of the output item
      Parameters:
        `mapName` - name of the `IMap`
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### mapUsingIMap

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a mapping stage where for each item a lookup in the supplied `IMap` using the grouping key 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:

           V value = map.get(groupingKey);
           return mapFn.apply(item, value);

      Sample usage:

           items.groupingKey(Item::getDetailId)
                .mapUsingIMap(enrichingMap, (item, detail) -> item.setDetail(detail));

      This stage is similar to `stageWithoutKey.mapUsingIMap()`, but here Jet knows the key and uses it to partition and distribute the input in order to achieve data locality. The value it fetches from the `IMap` is stored on the cluster member where the processing takes place. However, if the map doesn't use the default partitioning strategy, data locality will be broken.
      Specified by:
        `mapUsingIMap` in interface `GeneralStageWithKey<T,K>`
      Type Parameters:
        `V` - type of the value in the `IMap`
        `R` - type of the output item
      Parameters:
        `iMap` - the `IMap` to use as the context
        `mapFn` - the mapping function
      Returns:
        the newly attached stage

    - #### mapUsingContext

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a mapping stage which applies the given 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.

      Jet uses the key-extracting function specified on this stage for partitioning: all the items with the same key will see the same context instance (but note that the same instance serves many keys). One case where this is useful is fetching data from an external system because you can use a near-cache without duplicating the cached data.

      Sample usage:

           items.groupingKey(Item::getDetailId)
                .mapUsingContext(
                    ContextFactory.withCreateFn(jet -> new ItemDetailRegistry()),
                    (reg, key, item) -> item.setDetail(reg.fetchDetail(key))
                );

      ### 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 `GeneralStageWithKey<T,K>`
      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
                                                             TriFunction<? super C,? super K,? super T,CompletableFuture<R>> mapAsyncFn)

      Description copied from interface: `GeneralStageWithKey`
      Asynchronous version of `GeneralStageWithKey.mapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.TriFunction<? super C, ? super K, ? 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.

      Sample usage:

           items.groupingKey(Item::getDetailId)
                .mapUsingContextAsync(
                    ContextFactory.withCreateFn(jet -> new ItemDetailRegistry()),
                    (reg, key, item) -> reg.fetchDetailAsync(key)
                                           .thenApply(detail -> item.setDetail(detail))
                );

      The latency of the async call will add to the total latency of the output.
      Specified by:
        `mapUsingContextAsync` in interface `GeneralStageWithKey<T,K>`
      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
                                                         TriPredicate<? super C,? super K,? super T> filterFn)

      Description copied from interface: `GeneralStageWithKey`
      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`.

      Jet uses the key-extracting function specified on this stage for partitioning: all the items with the same key will see the same context instance (but note that the same instance serves many keys). One case where this is useful is fetching data from an external system because you can use a near-cache without duplicating the cached data.

      Sample usage:

           items.groupingKey(Item::getDetailId)
                .filterUsingContext(
                    ContextFactory.withCreateFn(jet -> new ItemDetailRegistry()),
                    (reg, key, item) -> reg.fetchDetail(key).contains("blade")
                );

      ### 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 `GeneralStageWithKey<T,K>`
      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
                                                              TriFunction<? super C,? super K,? super T,CompletableFuture<Boolean>> filterAsyncFn)

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

      The function must not return a null future.

      Sample usage:

           items.groupingKey(Item::getDetailId)
                .filterUsingContextAsync(
                    ContextFactory.withCreateFn(jet -> new ItemDetailRegistry()),
                    (reg, key, item) -> reg.fetchDetailAsync(key)
                                           .thenApply(detail -> detail.contains("blade"))
                );

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

      Specified by:
        `filterUsingContextAsync` in interface `GeneralStageWithKey<T,K>`
      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
                                                            TriFunction<? super C,? super K,? super T,? extends Traverser<? extends R>> flatMapFn)

      Description copied from interface: `GeneralStageWithKey`
      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 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`.

      Jet uses the key-extracting function specified on this stage for partitioning: all the items with the same key will see the same context instance (but note that the same instance serves many keys). One case where this is useful is fetching data from an external system because you can use a near-cache without duplicating the cached data.

      Sample usage:

           StreamStage<Part> parts = products
               .groupingKey(Product::getId)
               .flatMapUsingContext(
                   ContextFactory.withCreateFn(jet -> new PartRegistry()),
                   (registry, productId, product) -> Traversers.traverseIterable(
                           registry.fetchParts(productId))
               );

      ### 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 `GeneralStageWithKey<T,K>`
      Type Parameters:
        `C` - type of context object
        `R` - type of the output items
      Parameters:
        `contextFactory` - the context factory
        `flatMapFn` - a stateless flatmapping function. 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
                                                                 TriFunction<? super C,? super K,? super T,CompletableFuture<Traverser<R>>> flatMapAsyncFn)

      Description copied from interface: `GeneralStageWithKey`
      Asynchronous version of `GeneralStageWithKey.flatMapUsingContext(com.hazelcast.jet.pipeline.ContextFactory<C>, com.hazelcast.jet.function.TriFunction<? super C, ? super K, ? super T, ? extends com.hazelcast.jet.Traverser<? extends 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.

      Sample usage:

           StreamStage<Part> productParts = products
               .groupingKey(Product::getId)
               .flatMapUsingContextAsync(
                   ContextFactory.withCreateFn(jet -> new PartRegistry()),
                   (registry, productId, product) -> registry
                           .fetchPartsAsync(productId)
                           .thenApply(parts -> Traversers.traverseIterable(parts))
               );

      The latency of the async call will add to the latency of the items.

      Specified by:
        `flatMapUsingContextAsync` in interface `GeneralStageWithKey<T,K>`
      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), but the future must not return null traverser, but can return an empty traverser.
      Returns:
        the newly attached stage

    - #### customTransform

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s. The inbound edge will be distributed and partitioned using the key function assigned to this stage.

      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 `GeneralStageWithKey<T,K>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors

    - #### customTransform

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s. The inbound edge will be distributed and partitioned using the key function assigned to this stage.

      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 `GeneralStageWithKey<T,K>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors

    - #### customTransform

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

      Description copied from interface: `GeneralStageWithKey`
      Attaches a stage with a custom transform based on the provided supplier of Core API `Processor`s. The inbound edge will be distributed and partitioned using the key function assigned to this stage.

      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 `GeneralStageWithKey<T,K>`
      Type Parameters:
        `R` - the type of the output items
      Parameters:
        `stageName` - a human-readable name for the custom stage
        `procSupplier` - the supplier of processors

Copyright © 2019. All rights reserved.