Flink – Trigger，Evictor

public abstract class Trigger<T, W extends Window> implements Serializable {

    /**
     * Called for every element that gets added to a pane. The result of this will determine
     * whether the pane is evaluated to emit results.
     *
     * @param element The element that arrived.
     * @param timestamp The timestamp of the element that arrived.
     * @param window The window to which the element is being added.
     * @param ctx A context object that can be used to register timer callbacks.
     */
    public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;

    /**
     * Called when a processing-time timer that was set using the trigger context fires.
     *
     * @param time The timestamp at which the timer fired.
     * @param window The window for which the timer fired.
     * @param ctx A context object that can be used to register timer callbacks.
     */
    public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;

    /**
     * Called when an event-time timer that was set using the trigger context fires.
     *
     * @param time The timestamp at which the timer fired.
     * @param window The window for which the timer fired.
     * @param ctx A context object that can be used to register timer callbacks.
     */
    public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;

    /**
     * Called when several windows have been merged into one window by the
     * {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}.
     *
     * @param window The new window that results from the merge.
     * @param ctx A context object that can be used to register timer callbacks and access state.
     */
    public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
        throw new RuntimeException("This trigger does not support merging.");
    }

/**
 * A {@link Trigger} that fires once the current system time passes the end of the window
 * to which a pane belongs.
 */
public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow> {
    private static final long serialVersionUID = 1L;
    
    private ProcessingTimeTrigger() {}
    
    @Override
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
        ctx.registerProcessingTimeTimer(window.maxTimestamp()); //對於processingTime，element的trigger時間是current+window，所以這裏需要注冊定時器去觸發
        return TriggerResult.CONTINUE;
    }
    
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }
    
    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {//觸發後調用
        return TriggerResult.FIRE_AND_PURGE;
    }
    
    @Override
    public String toString() {
        return "ProcessingTimeTrigger()";
    }
    
    /**
    * Creates a new trigger that fires once system time passes the end of the window.
    */
    public static ProcessingTimeTrigger create() {
        return new ProcessingTimeTrigger();
    }
}

public class CountTrigger<W extends Window> extends Trigger<Object, W> {

    private final long maxCount;

    private final ReducingStateDescriptor<Long> stateDesc =
            new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);

    private CountTrigger(long maxCount) {
        this.maxCount = maxCount;
    }

    @Override
    public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
        ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //從backend取出conunt state
        count.add(1L); //加1
        if (count.get() >= maxCount) {
            count.clear();
            return TriggerResult.FIRE;
        }
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
        ctx.mergePartitionedState(stateDesc); //先調用merge，底層backend裏麵的window進行merge
        ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //merge後再取出state，count，進行判斷
        if (count.get() >= maxCount) {
            return TriggerResult.FIRE;
        }
        return TriggerResult.CONTINUE;
    }

enum TriggerResult {
    CONTINUE(false, false), FIRE_AND_PURGE(true, true), FIRE(true, false), PURGE(false, true);
    
    private final boolean fire;
    private final boolean purge;
}

/**
 * Merges two {@code TriggerResults}. This specifies what should happen if we have
 * two results from a Trigger, for example as a result from
 * {@link Trigger#onElement(Object, long, Window, Trigger.TriggerContext)} and
 * {@link Trigger#onEventTime(long, Window, Trigger.TriggerContext)}.
 *
 * <p>
 * For example, if one result says {@code CONTINUE} while the other says {@code FIRE}
 * then {@code FIRE} is the combined result;
 */
public static TriggerResult merge(TriggerResult a, TriggerResult b) {
    if (a.purge || b.purge) {
        if (a.fire || b.fire) {
            return FIRE_AND_PURGE;
        } else {
            return PURGE;
        }
    } else if (a.fire || b.fire) {
        return FIRE;
    } else {
        return CONTINUE;
    }
}

/**
     * A context object that is given to {@link Trigger} methods to allow them to register timer
     * callbacks and deal with state.
     */
    public interface TriggerContext {

        long getCurrentProcessingTime();
        long getCurrentWatermark();
    
        /**
         * Register a system time callback. When the current system time passes the specified
         * time {@link Trigger#onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.
         *
         * @param time The time at which to invoke {@link Trigger#onProcessingTime(long, Window, TriggerContext)}
         */
        void registerProcessingTimeTimer(long time);
        void registerEventTimeTimer(long time);
    
        void deleteProcessingTimeTimer(long time);
        void deleteEventTimeTimer(long time);
    

        <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
    }

public interface OnMergeContext extends TriggerContext {
    <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor);
}

/**
 * {@code Context} is a utility for handling {@code Trigger} invocations. It can be reused
 * by setting the {@code key} and {@code window} fields. No internal state must be kept in
 * the {@code Context}
 */
public class Context implements Trigger.OnMergeContext {
    protected K key; //Context對應的window上下文
    protected W window;

    protected Collection<W> mergedWindows; //onMerge中被賦值

    @SuppressWarnings("unchecked")
    public <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor) {
        try {
            return WindowOperator.this.getPartitionedState(window, windowSerializer, stateDescriptor); //從backend裏麵讀出改window的狀態，即window buffer
        } catch (Exception e) {
            throw new RuntimeException("Could not retrieve state", e);
        }
    }

    @Override
    public <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor) {
        if (mergedWindows != null && mergedWindows.size() > 0) {
            try {
                WindowOperator.this.getStateBackend().mergePartitionedStates(window, //在backend層麵把mergedWindows merge到window中
                        mergedWindows,
                        windowSerializer,
                        stateDescriptor);
            } catch (Exception e) {
                throw new RuntimeException("Error while merging state.", e);
            }
        }
    }

    @Override
    public void registerProcessingTimeTimer(long time) {
        Timer<K, W> timer = new Timer<>(time, key, window);
        // make sure we only put one timer per key into the queue
        if (processingTimeTimers.add(timer)) {
            processingTimeTimersQueue.add(timer);
            //If this is the first timer added for this timestamp register a TriggerTask
            if (processingTimeTimerTimestamps.add(time, 1) == 0) { //如果這個window是第一次注冊的話
                ScheduledFuture<?> scheduledFuture = WindowOperator.this.registerTimer(time, WindowOperator.this); //對於processTime必須注冊定時器主動觸發
                processingTimeTimerFutures.put(time, scheduledFuture);
            }
        }
    }

    @Override
    public void registerEventTimeTimer(long time) {
        Timer<K, W> timer = new Timer<>(time, key, window);
        if (watermarkTimers.add(timer)) {
            watermarkTimersQueue.add(timer);
        }
    }

    //封裝一遍trigger的接口，並把self作為context傳入trigger的接口中
    public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
        return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
    }

    public TriggerResult onProcessingTime(long time) throws Exception {
        return trigger.onProcessingTime(time, window, this);
    }

    public TriggerResult onEventTime(long time) throws Exception {
        return trigger.onEventTime(time, window, this);
    }

    public TriggerResult onMerge(Collection<W> mergedWindows) throws Exception {
        this.mergedWindows = mergedWindows;
        return trigger.onMerge(window, this);
    }

}

/**
 * An {@code Evictor} can remove elements from a pane before it is being processed and after
 * window evaluation was triggered by a
 * {@link org.apache.flink.streaming.api.windowing.triggers.Trigger}.
 *
 * <p>
 * A pane is the bucket of elements that have the same key (assigned by the
 * {@link org.apache.flink.api.java.functions.KeySelector}) and same {@link Window}. An element can
 * be in multiple panes of it was assigned to multiple windows by the
 * {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. These panes all
 * have their own instance of the {@code Evictor}.
 *
 * @param <T> The type of elements that this {@code Evictor} can evict.
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
 */
public interface Evictor<T, W extends Window> extends Serializable {

    /**
     * Computes how many elements should be removed from the pane. The result specifies how
     * many elements should be removed from the beginning.
     *
     * @param elements The elements currently in the pane.
     * @param size The current number of elements in the pane.
     * @param window The {@link Window}
     */
    int evict(Iterable<StreamRecord<T>> elements, int size, W window);
}

/**
 * An {@link Evictor} that keeps only a certain amount of elements.
 *
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
 */
public class CountEvictor<W extends Window> implements Evictor<Object, W> {
    private static final long serialVersionUID = 1L;
    
    private final long maxCount;
    
    private CountEvictor(long count) {
        this.maxCount = count;
    }
    
    @Override
    public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {
        if (size > maxCount) {
            return (int) (size - maxCount);
        } else {
            return 0;
        }
    }
    
    /**
    * Creates a {@code CountEvictor} that keeps the given number of elements.
    *
    * @param maxCount The number of elements to keep in the pane.
    */
    public static <W extends Window> CountEvictor<W> of(long maxCount) {
        return new CountEvictor<>(maxCount);
    }
}

/**
 * An {@link Evictor} that keeps elements for a certain amount of time. Elements older
 * than {@code current_time - keep_time} are evicted.
 *
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.
 */
public class TimeEvictor<W extends Window> implements Evictor<Object, W> {
    private static final long serialVersionUID = 1L;
    
    private final long windowSize;
    
    public TimeEvictor(long windowSize) {
        this.windowSize = windowSize;
    }
    
    @Override
    public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {
        int toEvict = 0;
        long currentTime = Iterables.getLast(elements).getTimestamp();
        long evictCutoff = currentTime - windowSize;
        for (StreamRecord<Object> record: elements) {
            if (record.getTimestamp() > evictCutoff) {
                break;
            }
            toEvict++;
        }
        return toEvict;
    }
}