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

On restart it can happen that partition1 has one very recent event and partition2 has one old event. If partition1 is polled first and the event is emitted, it will advance the watermark. Then later partition2 is polled and its event might be dropped as late. This utility helps you track watermarks per partition and decide when to emit it.

2. Some partition having no data

It can happen that some partition does not have any events at all and others do. Or that the processor is not assigned any external partitions. In this both cases no watermarks will be emitted. This utility supports idle timeout: if some partition does not have any event during this time, it will be marked as idle. If all partitions are idle or there are no partitions, special idle message will be emitted and the downstream will exclude this processor from watermark coalescing.

Usage

API is designed to be used as a flat-mapping step in Traverser. Your source might follow this pattern:

   public boolean complete() {
       if (traverser == null) {
           List<Record> records = poll(); // get a batch of events from external source
           if (records.isEmpty()) {
               traverser = watermarkSourceUtil.handleNoEvent();
           } else {
               traverser = traverserIterable(records)
                   .flatMap(event -> watermarkSourceUtil.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