com.hazelcast.jet.core

## Class EventTimeMapper<T>

- java.lang.Object
- - com.hazelcast.jet.core.EventTimeMapper<T>

- Type Parameters:
    `T` - the event type

      public class EventTimeMapper<T>
      extends Object

  A utility to that helps a source emit events according to a given `EventTimePolicy`. Generally this class should be used if a source needs to emit `watermarks`. The mapper deals with the following concerns:

  #### 1. Reading partition by partition

  Upon restart it can happen that partition partition1 has one very recent event and partition2 has an old one. If we poll partition1 first and emit its recent event, it will advance the watermark. When we poll partition2 later on, its event will be behind the watermark and can be dropped as late. This utility tracks the event timestamps for each source partition individually and allows the processor to emit the watermark that is correct with respect to all the partitions.

  #### 2. Some partition having no data

  It can happen that some partition does not have any events at all while others do or the processor doesn't get any external partitions assigned to it. If we simply wait for the timestamps in all partitions to advance to some point, we won't be emitting any watermarks. This utility supports the idle timeout: if there's no new data from a partition after the timeout elapses, it will be marked as idle, allowing the processor's watermark to advance as if that partition didn't exist. If all partitions are idle or there are no partitions, the processor will emit a special idle message and the downstream will exclude this processor from watermark coalescing.

  #### 3. Wrapping of events

  Events may need to be wrapped with the extracted timestamp if `EventTimePolicy.wrapFn()` is set.

  #### 4. Throttling of Watermarks

  Watermarks are only consumed by windowing operations and emitting watermarks more frequently than the given `EventTimePolicy.watermarkThrottlingFrameSize()` is wasteful since they are broadcast to all processors. The mapper ensures that watermarks are emitted according to the throttling frame size.

  ### Usage

  The API is designed to be used as a flat-mapping step in the `Traverser` that holds the output data. Your source can follow this pattern:

       public boolean complete() {
           if (traverser == null) {
               List<Record> records = poll();
               if (records.isEmpty()) {
                   traverser = eventTimeMapper.flatMapIdle();
               } else {
                   traverser = traverseIterable(records)
                       .flatMap(event -> eventTimeMapper.flatMapEvent(
                            event, event.getPartition()));
               }
               traverser = traverser.onFirstNull(() -> traverser = null);
           }
           emitFromTraverser(traverser, event -> {
               if (!(event instanceof Watermark)) {
                   // store your offset after event was emitted
                   offsetsMap.put(event.getPartition(), event.getOffset());
               }
           });
           return false;
       }

  Other methods:
  - Call `addPartitions(int)` and `removePartition(int)` to change your partition count initially or whenever the count changes.
  - If you support state snapshots, save the value returned by `getWatermark(int)` for all partitions to the snapshot. When restoring the state, call `restoreWatermark(int, long)`.
    You should save the value under your external partition key so that the watermark value can be restored to correct processor instance. The key should also be wrapped using `broadcastKey()`, because the external partitions don't match Hazelcast partitions. This way, all processor instances will see all keys and they can restore the partitions they handle and ignore others.
  Since:
    3.0

- - ### Field Summary

    Fields
    Modifier and Type | Field and Description
    --- | ---
    `static long` | `NO_NATIVE_TIME` Value to use as the `nativeEventTime` argument when calling `flatMapEvent(Object, int, long)` when there's no native event time to supply.

  - ### Constructor Summary

    Constructors
    Constructor and Description
    `EventTimeMapper(EventTimePolicy<? super T> eventTimePolicy)` The partition count is initially set to 0, call `addPartitions(int)` to add partitions.

  - ### Method Summary

    All Methods Instance Methods Concrete Methods
    Modifier and Type | Method and Description
    --- | ---
    `void` | `addPartitions(int addedCount)` Adds `addedCount` partitions.
    `Traverser<Object>` | `flatMapEvent(T event, int partitionIndex, long nativeEventTime)` Flat-maps the given `event` by (possibly) prepending it with a watermark.
    `Traverser<Object>` | `flatMapIdle()` Call this method when there is no event coming.
    `long` | `getWatermark(int partitionIndex)` Watermark value to be saved to state snapshot for the given source partition index.
    `int` | `partitionCount()` Returns the current partition count.
    `Traverser<Object>` | `removePartition(int partitionIndex)` Removes a partition that will no longer have events.
    `void` | `restoreWatermark(int partitionIndex, long wm)` Restore watermark value from state snapshot.

    - ### Methods inherited from class java.lang.Object

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

- - ### Field Detail

    - #### NO_NATIVE_TIME

          public static final long NO_NATIVE_TIME

      Value to use as the `nativeEventTime` argument when calling `flatMapEvent(Object, int, long)` when there's no native event time to supply.
      See Also:
        Constant Field Values

  - ### Constructor Detail

    - #### EventTimeMapper

          public EventTimeMapper(EventTimePolicy<? super T> eventTimePolicy)

      The partition count is initially set to 0, call `addPartitions(int)` to add partitions.
      Parameters:
        `eventTimePolicy` - event time policy as passed in `Sources.streamFromProcessorWithWatermarks(java.lang.String, java.util.function.Function<com.hazelcast.jet.core.EventTimePolicy<? super T>, com.hazelcast.jet.core.ProcessorMetaSupplier>, boolean)`

  - ### Method Detail

    - #### flatMapEvent

          @Nonnull
          public Traverser<Object> flatMapEvent(T event,
                                                         int partitionIndex,
                                                         long nativeEventTime)

      Flat-maps the given `event` by (possibly) prepending it with a watermark. Designed to use when emitting from traverser:

               Traverser t = traverserIterable(...)
                   .flatMap(event -> eventTimeMapper.flatMapEvent(
                           event, event.getPartition(), nativeEventTime));

      Parameters:
        `event` - the event
        `partitionIndex` - the source partition index the event came from
        `nativeEventTime` - native event time in case no `timestampFn` was supplied or `NO_NATIVE_TIME` if the event has no native timestamp

    - #### flatMapIdle

          @Nonnull
          public Traverser<Object> flatMapIdle()

      Call this method when there is no event coming. It returns a traverser with 0 or 1 object (the watermark).

    - #### addPartitions

          public void addPartitions(int addedCount)

      Adds `addedCount` partitions. Added partitions will be initially considered active and having watermark value equal to the last emitted watermark. Their indices will follow current highest index.

      You can call this method whenever new partitions are detected.

      Parameters:
        `addedCount` - number of added partitions, must be >= 0

    - #### removePartition

          public Traverser<Object> removePartition(int partitionIndex)

      Removes a partition that will no longer have events. If we were waiting for a watermark from it, the returned traverser might contain a watermark to emit.

      Note that the indexes of partitions with index larger than the given `partitionIndex` will be decremented by 1.

      Parameters:
        `partitionIndex` - the index of the removed partition

    - #### partitionCount

          public int partitionCount()

      Returns the current partition count.

    - #### getWatermark

          public long getWatermark(int partitionIndex)

      Watermark value to be saved to state snapshot for the given source partition index. The returned value should be `restored` to a processor handling the same partition after restart.

      Method is meant to be used from `Processor.saveToSnapshot()`.

      Parameters:
        `partitionIndex` - 0-based source partition index.
      Returns:
        A value to save to state snapshot

    - #### restoreWatermark

          public void restoreWatermark(int partitionIndex,
                                       long wm)

      Restore watermark value from state snapshot.

      Method is meant to be used from `Processor.restoreFromSnapshot(Inbox)`.

      See `getWatermark(int)`.

      Parameters:
        `partitionIndex` - 0-based source partition index.
        `wm` - watermark value to restore

Copyright © 2019. All rights reserved.