How to implement a flink countTriggerWithTimeout with dynamic timeouts for each incoming element

Quite new to flink stream processing. Here is my requirement:
Alert the user when 2 or more elements were received in the last 20 seconds. If less than 2 elements were received in 20 seconds dont alert, just restart the counting and time.
The count and interval varies for each element.

Here's my code:

dataStream
.keyBy("id")
.window(EventTimeSessionWindows.withDynamicGap((event) -> event.getThresholdInterval()))
.trigger(new CountTriggerWithTimeout<TimeWindow>())

TriggerCode:
public class CountTriggerWithTimeout<W extends TimeWindow> extends Trigger<SystemEvent, W> {

  private ReducingStateDescriptor<Long> countState =
      new ReducingStateDescriptor<Long>("count", new Sum(), LongSerializer.INSTANCE);
  private ReducingStateDescriptor<Long> processedState =
      new ReducingStateDescriptor<Long>("processed", new Sum(), LongSerializer.INSTANCE);

  @Override
  public TriggerResult onElement(SystemEvent element, long timestamp, W window, TriggerContext ctx)
      throws Exception {
    ReducingState<Long> count = ctx.getPartitionedState(countState);
    ReducingState<Long> processed = ctx.getPartitionedState(processedState);
    count.add(1L);
    processed.add(0L);
    if (count.get() >= element.getThresholdCount() && processed.get() == 0) {
      processed.add(1L);
      return TriggerResult.FIRE_AND_PURGE;
    }
    if (timestamp >= window.getEnd()) { 
      return TriggerResult.PURGE;
    }
    return TriggerResult.CONTINUE;
  }

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

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

  @Override
  public void clear(W window, TriggerContext ctx) throws Exception {
    ctx.getPartitionedState(countState).clear();
    ctx.getPartitionedState(processedState).clear();
  }
  
  @Override
  public boolean canMerge() {
    return true;
  }

  class Sum implements ReduceFunction<java.lang.Long> {
    @Override
    public Long reduce(Long value1, Long value2) throws Exception {
      return value1 + value2;
    }
  }
}

Earlier when I was using

dataStream
.timeWindow(Time.seconds(1))
.trigger(new CountTriggerWithTimeout<TimeWindow>())

everything was working perfectly fine. Since there is a requirement to read the window time from element, I started using EventTimeSessionWindow and added canMerge() function in the trigger. Since then, nothing is working. clear() is not getting invoked ever, nor are onProcessingTime() & onEventTime(). I see that timestamp is always set to the same value, irrespective of when the element was received.

My requirement is to "fire & purge" when count >= threshold within event.getThresholdInterval(). If count < threshold within event.getThresholdInterval() then purge i.e. invoke clear to clear the count and state and restart. Is there a way to achieve this with timeWindow instead of EventTimeSessionWindows?

Please help me fix this issue.

Thanks...

Why dont you use a simple Tumbling Windows of 20 seconds and count the elements on it:

source
		.keyBy("id")
		.timeWindow(Time.seconds(20))
		.process(new ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple, TimeWindow>() {
			@Override
			public void process(Tuple key, ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple, TimeWindow>.Context ctx,
					Iterable<Tuple2<String, Integer>> in, Collector<String> out) throws Exception {

				if (Lists.newArrayList(in).size() >= 2) {
					out.collect("Two or more elements between "
							+ Instant.ofEpochMilli(ctx.window().getStart())
							+ " " + Instant.ofEpochMilli(ctx.window().getEnd()));
				}
			}
		})