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 Watermarks 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

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