twalthr commented on code in PR #27935:
URL: https://github.com/apache/flink/pull/27935#discussion_r3092698461
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java:
##########
@@ -154,6 +154,12 @@ public void processWatermark(Watermark mark) throws
Exception {
@Override
public void onEventTime(InternalTimer<RowData, Object> timer) throws
Exception {
final Object namedTimer = timer.getNamespace();
+ // Remove the fired timer's state entry immediately to prevent stale
entries from
+ // accumulating. Without this, entries for fired timers would persist
until the same
+ // timer name is re-registered or the state is explicitly cleared.
+ if (namedTimersMapState != null && namedTimer !=
VoidNamespace.INSTANCE) {
+ namedTimersMapState.remove((StringData) namedTimer);
+ }
processTableRunner.ingestTimerEvent(
timer.getKey(),
namedTimer == VoidNamespace.INSTANCE ? null : (StringData)
namedTimer,
Review Comment:
good catch! some nit comments
```suggestion
boolean isNamedTimer = namedTimer != VoidNamespace.INSTANCE;
// Remove the fired timer's state entry immediately to prevent stale
entries from
// accumulating. Without this, entries for fired timers would
persist until the same
// timer name is re-registered or the state is explicitly cleared.
if (isNamedTimer) {
namedTimersMapState.remove((StringData) namedTimer);
}
processTableRunner.ingestTimerEvent(
timer.getKey(),
isNamedTimer ? (StringData) namedTimer: null,
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java:
##########
@@ -664,7 +667,12 @@ public static class OptionalOnTimeFunction extends
AppendProcessTableFunctionBas
public void eval(Context ctx, @ArgumentHint(SET_SEMANTIC_TABLE) Row r)
{
final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
collectEvalEvent(timeCtx, r);
- collectCreateTimer(timeCtx, "t", 2);
+ Long wm = timeCtx.currentWatermark();
+ // Register at wm+1 to always target the immediate next watermark:
the timer fires
+ // exactly once per watermark advance, and each new row
re-registers the timer for the
+ // following watermark step, demonstrating repeated timer
re-registration.
+ long timer = wm == null || wm < 0 ? 1 : wm + 1;
Review Comment:
wouldn't this be more correct and allowing for watermarks before epoch.
```suggestion
long timer = wm == null ? Long.MIN_VALUE + 1: wm + 1;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
