com.hazelcast.jet.core

## Class WatermarkSourceUtil<T>

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

- Type Parameters:
    `T` - event type

      public class WatermarkSourceUtil<T>
      extends Object

  A utility to help emitting `Watermark`s from a source which reads events from multiple external partitions.

  ### The problems

  #### 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 watermork, to be dropped as late. This utility tracks the event timestamps for each 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.

  ### 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 = wmSourceUtil.handleNoEvent();
               } else {
                   traverser = traverserIterable(records)
                       .flatMap(event -> wmSourceUtil.handleEvent(
                            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 `increasePartitionCount(int)` to set your partition count initially or whenever the count increases.
  - 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 partition they handle and ignore others.

- - ### Constructor Summary

    Constructors
    Constructor and Description
    `WatermarkSourceUtil(WatermarkGenerationParams<? super T> params)` A constructor.

  - ### Method Summary

    All Methods Instance Methods Concrete Methods
    Modifier and Type | Method and Description
    --- | ---
    `long` | `getWatermark(int partitionIndex)` Watermark value to be saved to state snapshot for the given source partition index.
    `Traverser<Object>` | `handleEvent(T event, int partitionIndex)` Flat-maps the given `event` by (possibly) prepending it with a watermark.
    `Traverser<Object>` | `handleNoEvent()` Call this method when there is no event coming.
    `void` | `increasePartitionCount(int newPartitionCount)` Changes the partition count.
    `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`

- - ### Constructor Detail

    - #### WatermarkSourceUtil

          public WatermarkSourceUtil(WatermarkGenerationParams<? super T> params)

      A constructor.

      The partition count is initially set to 0, call `increasePartitionCount(int)` to set it.

  - ### Method Detail

    - #### handleEvent

          @Nonnull
          public Traverser<Object> handleEvent(T event,
                                                        int partitionIndex)

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

               Traverser t = traverserIterable(...)
                   .flatMap(event -> watermarkSourceUtil.flatMap(event, event.getPartition()));

    - #### handleNoEvent

          @Nonnull
          public Traverser<Object> handleNoEvent()

      Call this method when there is no event coming. It returns a traverser with 0 or 1 object (the watermark). If you need just the Watermark, call `next()` on the result.

    - #### increasePartitionCount

          public void increasePartitionCount(int newPartitionCount)

      Changes the partition count. The new partition count must be higher or equal to the current count.

      You can call this method at any moment. Added partitions will be considered active initially.

      Parameters:
        `newPartitionCount` - partition count, must be higher than the current 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 © 2018. All rights reserved.