xljtswf created FLINK-30208: ------------------------------- Summary: avoid unconditional state update in CountTrigger#onElement Key: FLINK-30208 URL: https://issues.apache.org/jira/browse/FLINK-30208 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: xljtswf
In current CountTrigger#onElement, when one element is received, the state is updated unconditionally, and we then fetch the state again to check whether we need to clear the state. This implies we may update the state 2 times to process one element. I suppose to make following simplification: public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception { TriggerResult triggerResult; if (maxCount > 1) { ReducingState<Long> countState = ctx.getPartitionedState(stateDesc); Long currentCount = countState.get(); if (currentCount == null || currentCount < maxCount - 1) { countState.add(1L); triggerResult = TriggerResult.CONTINUE; } else { countState.clear(); triggerResult = TriggerResult.FIRE; } } else { triggerResult = TriggerResult.FIRE; } return triggerResult; } If this is approved, I will make a pr then. Thanks! -- This message was sent by Atlassian Jira (v8.20.10#820010)