Flink 自定义 countAndTimeTrigger

/**
 * @author shengjk1
 * @date 2019/9/4
 */
public class CountAndTimeTrigger extends Trigger<Object, TimeWindow> {
	private static final long serialVersionUID = 1L;
	
	private final long maxCount;
	
	private final ReducingStateDescriptor<Long> stateDesc =
			new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);
	
	public CountAndTimeTrigger(long maxCount) {
		super();
		this.maxCount = maxCount;
	}
	
	
	@Override
	public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
		ctx.registerProcessingTimeTimer(window.maxTimestamp());
		
		ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
		count.add(1L);
		if (count.get() >= maxCount) {
			count.clear();
			return TriggerResult.FIRE_AND_PURGE;
		}
		return TriggerResult.CONTINUE;
	}
	
	@Override
	public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
		return TriggerResult.FIRE;
	}
	
	@Override
	public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
		return TriggerResult.CONTINUE;
	}
	
	@Override
	public boolean canMerge() {
		return false;
	}
	
	@Override
	public void onMerge(TimeWindow window, OnMergeContext ctx) throws Exception {
		ctx.mergePartitionedState(stateDesc);
		long windowMaxTimestamp = window.maxTimestamp();
		if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
			ctx.registerProcessingTimeTimer(windowMaxTimestamp);
		}
	}
	
	@Override
	public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
		ctx.getPartitionedState(stateDesc).clear();
		ctx.deleteProcessingTimeTimer(window.maxTimestamp());
	}
	
	
	private static class Sum implements ReduceFunction<Long> {
		private static final long serialVersionUID = 1L;
		
		@Override
		public Long reduce(Long value1, Long value2) throws Exception {
			return value1 + value2;
		}
	}
}