Trigger and Evictor in Flink

// UserEvent là kiểu dữ liệu của sự kiện người dùng
// TimeWindow là loại cửa sổ theo thời gian
public class CustomCountTrigger extends Trigger<UserEvent, TimeWindow> {

    // Xác định worker nào đang làm việc.
    private static final long serialVersionUID = 1L;

    // ValueState to count enlements in the window
    private final ValueStateDescriptor<Integer> countStateDesc = new ValueStateDescriptor<>("count", Integer.class);

    @Override
    public TriggerResult onElement(UserEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        // Get or create the count state
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        int count = countState.value();
        if (count == null) {
            count = 0;
        }

        // Increment the count
        count++;
        countState.update(count);

        // if this is the first element, register an event-time for end-of-window.
        // (Timers are set window end timestamp, so when the watermatk passes windown.end, onEventTime will fire)

        if (count == 1) {
            long windowEnd = window.getEnd(); // Get the end timestamp of the window
            ctx.registerEventTimeTimer(windowEnd);
        }

        if (count >= 5) {
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long timestamp, TimeWindow window, TriggerContext triggerContext) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        // This is called when the event-time timer fires (i.e when the watermark passes the end of the window)
        if (timestamp == window.getEnd()){
            return TriggerResult.FIRE_AND_PURGE;
        }

        // if it's not the end of the window, continue processing
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        // Clear the count state when the window is purged/closed.
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        countState.clear();
        // we don't need to clean up timers because they will be cleared automatically
    }
    
}

public class SesionEndTrigger extends Trigger<UserEvent, TimeWindow>{
    private static final long serialVersionUID = 1L;

    @Override
    public TriggerResult onElement(UserEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        // Register a processing-time timer for the end of the session
        ctx.registerProcessingTimeTimer(window.getEnd());

        if ("LOGOUT".equals(element.geEventType()) || "CHECKOUT_COMPLETE".equals(element.getEventType())){
            return TriggerResult.FIRE_AND_PURGE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

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

    @Override
    public TriggerResult onEventTime(long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        // round the timestamp to the next millisecond
        if (timestamp + 1 == window.getEnd()){
            return TriggerResult.FIRE_AND_PURGE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public void onMerge(TimeWindow window, OnMergeContext ctx) throws Exception {
        // When windows merge, the new window will have the latest end time
        ctx.registerProcessingTimeTimer(window.getEnd());
    }

    @Override
    public boolean canMerge() {
        return true;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }
}