feat: Implement idle source handling for WatermarkStatus

[lgbo-ustc]

Summary

Implements idle source handling for Gluten Flink (issue apache/gluten#12340), using native SystemProcessingTimeScheduler timer + NativeCallbackBridge->onProcessingTime to detect idle input and emit WatermarkStatus.IDLE/ACTIVE.

Changes

New files

• WatermarkIdleTracker.h/.cpp — Pure-logic idle detection class (onRecord, checkIdle, isIdle, isEnabled)
• tests/WatermarkIdleTrackerTest.cpp — 9 unit tests

Modified files

• WatermarkAssigner.h/.cpp — Integrate idleTracker + SystemProcessingTimeScheduler timer + callbackBridge->onProcessingTime; emit ACTIVE on data arrival, IDLE via checkWatermarkStatus() hook
• WatermarkGenerator.h/.cpp — Add WatermarkIdleTracker member, expose onRecord/checkIdle/idleTimeout for source-level use
• WatermarkSource.h/.cpp — Same timer+idle pattern as WatermarkAssigner for source-level watermark pushdown
• StatefulOperator.h/.cpp — Add virtual ~StatefulOperator() = default and virtual void checkWatermarkStatus(int64_t now) with recursive forwarding to targets
• StatefulTask.h/.cpp — Call checkWatermarkStatus() in next() after each drain attempt (both blocked and empty-output paths)
• CMakeLists.txt — Register new source files

Design

• Timer callback only sets std::atomic<bool> flag + calls bridge to wake Java mailbox — never writes to pendings_ directly
• Actual idle check + emitWatermarkStatus happens on driver thread in checkWatermarkStatus(), avoiding races
• checkWatermarkStatus is called on every drain attempt, so idle detection works even without callback bridge (e.g. WatermarkSource path)

Verification

• All 9 new WatermarkIdleTracker unit tests pass
• libvelox.so compiles successfully in Docker container lgbo_flink_native_compile
• velox4j and gluten-flink compile successfully with 63 existing tests passing

[lgbo-ustc]

Thanks for the PR! The design is clean and well-thought-out. Below are my detailed review comments:

---

WatermarkIdleTracker

• checkIdle with lastRecordWallTime_ == 0 baseline logic: When no record has ever arrived, checkIdle sets lastRecordWallTime_ = currentWallMs and returns false. This correctly prevents an immediate IDLE declaration on a cold source. However, if checkIdle is called repeatedly before any record arrives, each call will keep resetting lastRecordWallTime_ to the current time, effectively postponing idle detection indefinitely. Is this the intended behavior? In Flink, WatermarkAssignerOperator starts the idle timer from the first open() call, not from the first checkIdle(). Consider using a separate initialized_ flag to only set the baseline once.

• setIdle is public but only used in tests: The setIdle method is exposed in the public API but only used by tests. Consider making it protected and using friend class or FRIEND_TEST to limit its visibility. This avoids accidental misuse in production code.

• idle_ state transition asymmetry: onRecord transitions idle→active on the first record after idle, which is correct. But checkIdle only transitions active→idle after the timeout elapses. Consider whether a reset() method is needed for checkpoint/restore scenarios where the tracker state needs to be fully reset.

WatermarkAssigner / WatermarkSource

• scheduleIdleTimer race: timerPending_.exchange(true) is used to prevent duplicate timer registrations. However, there is a subtle race:

  1. Thread A calls scheduleIdleTimer, exchange(true) returns false, timer is registered.
  2. Timer fires on the scheduler thread, sets timerPending_ = false.
  3. Before checkWatermarkStatus runs on the driver thread, scheduleIdleTimer is called again, exchange(true) returns false, another timer is registered.
  4. The first timer's wake-up triggers a drain, then the second timer fires, triggering another drain.

  This is likely harmless (an extra wake-up), but the comment in WatermarkAssigner.h says timerPending_ "signals checkWatermarkStatus to perform an idle check." In reality, checkWatermarkStatus always performs the idle check regardless of timerPending_. The flag only guards against duplicate registrations. The comment should be clarified.

• Duplicate code between WatermarkAssigner and WatermarkSource: The idle timer management logic (scheduleIdleTimer, onIdleTimerFired, scheduler_ lifetime, timerPending_) is nearly identical between WatermarkAssigner and WatermarkSource. Consider extracting a shared helper class (e.g., IdleTimerManager) to avoid the duplication. This would also reduce the risk of fixing a bug in one place but not the other.

• ProcessingTimeScheduler header missing from includes: WatermarkAssigner.cpp and WatermarkSource.cpp use SystemProcessingTimeScheduler and ProcessingTimerTask but do not include the corresponding header directly. They rely on the header being transitively included via WatermarkAssigner.h / WatermarkSource.h. It is better practice to explicitly include the header in the .cpp files to make dependencies explicit.

• close() in destructor: ~WatermarkAssigner() calls WatermarkAssigner::close(), and ~WatermarkSource() calls WatermarkSource::close(). Calling virtual methods (or methods that themselves call virtual methods like close()) from constructors/destructors is generally discouraged in C++. While close() is not marked virtual in the diff, StatefulOperator::close() is called from it, and it could have virtual dispatch implications. Consider moving the scheduler cleanup to the destructor body directly or ensuring close() is non-virtual.

StatefulTask

• checkWatermarkStatus called on every drain attempt: In next(), checkWatermarkStatus is now called in three paths: when source is blocked, when output is empty, and when source is empty. The third path (sourceEmpty()) already returns nullptr without producing output, so calling checkWatermarkStatus there is correct. But note that when sourceEmpty() returns true and the scheduler is not running (no timer wake-up), checkWatermarkStatus is the only path for idle detection — which is the design intent and looks correct.

• Potential busy-loop when pendings_ becomes non-empty due to idle status: After checkWatermarkStatus emits a WatermarkStatus into pendings_, the flow falls through to popOutput(). This is correct. But if the idle timer fires frequently (e.g., a very short idleTimeout), the drain loop in next() could produce many WatermarkStatus events rapidly. Consider whether there should be a rate-limit or deduplication mechanism to avoid flooding the output with redundant status events.

Tests

• Good coverage: 9 tests cover the key scenarios: disabled, normal flow, idempotent idle, reactivation, early fire prevention, baseline, exact boundary, and explicit setIdle. Nicely done.

• Missing test for checkIdle baseline reset behavior: The test firstCheckSetsBaselineWithoutFiring only tests a single checkIdle call before a record. Consider adding a test that verifies the behavior when checkIdle is called multiple times before any record arrives (see the concern about the baseline logic above).

Minor / Nits

• WatermarkGenerator.h removes a TODO comment about idle source handling — great to see it implemented!
• WatermarkAssigner.h constructor init list has a whitespace-only change on the targets_ line (extra space). Consider reverting to keep the diff clean.
• The PR description references issue apache/gluten#12340 — consider adding Closes or Refs in the commit message for cross-repo linkage.

---

Overall, this is a solid implementation. The main architectural suggestion is to extract the shared idle timer logic to reduce duplication. The race condition is minor but the comment should be fixed. The checkIdle baseline reset behavior should be clarified or fixed.

[KevinyhZou]

Overall direction looks good: idle transitions on the driver thread + timer only waking the mailbox is the right split. The baseline reset issue is already fixed. Requesting changes on three blockers before merge (details in inline comments).

Blockers

1. Blocked path in StatefulTask::next() drops pending WatermarkStatus emitted by checkWatermarkStatus.
2. Idle timer is scheduled as now + idleTimeout instead of lastRecordWallTime + idleTimeout, so IDLE can be delayed by up to ~2× timeout.
3. After already IDLE, timers keep being re-registered → periodic no-op mailbox wakeups.

Also please

• Deduplicate idle-timer logic between WatermarkAssigner / WatermarkSource
• Avoid calling close() from destructor (extract scheduler shutdown)
• Add integration tests (not only WatermarkIdleTracker unit tests)
• Fix stale comments (idleCheckRequested_, “lazily created on first record”)

Nice parts: baselineInitialized_, driver-thread state transition, idempotent checkIdle, and timerPending_ duplicate-registration guard.

[KevinyhZou]

[Blocker 1] Blocked path drops already-emitted WatermarkStatus

if (future != nullptr && future->valid()) {
  checkWatermarkStatus(...);  // may push IDLE into pendings_
  return nullptr;             // discards that pending output
}

The empty-output path below correctly re-checks pendings_ and falls through to popOutput(). The blocked path does not.

Typical failure: timer wakes mailbox → next() → source still blocked → IDLE is computed this round but returned as nullptr.

Suggestion:

if (future != nullptr && future->valid()) {
  checkWatermarkStatus(TimeWindowUtil::getCurrentProcessingTime());
  if (!pendings_.empty()) {
    return popOutput();
  }
  return nullptr;
}

Please treat this as a merge blocker and add a regression test that covers blocked + idle emission in the same next() call.

[KevinyhZou]

[Blocker 2] Timer scheduled from now, not from last record time

int64_t fireAt = now + idleTimeout_;

checkIdle measures idleness from lastRecordWallTime_, but the wake-up timer is anchored at the current check time. Because addInput deliberately does not re-register the timer, a late check can push fire time close to lastRecord + 2 * idleTimeout, delaying IDLE by up to ~one full timeout.

Suggestion: expose last-record (or remaining) from WatermarkIdleTracker and schedule:

fireAt = lastRecordWallTime + idleTimeout;  // clamp to now if already due

---

[Blocker 3] Keep scheduling idle timers after already IDLE

scheduleIdleTimer(now) runs unconditionally even when idleTracker_ is already idle (checkIdle returns false and is a no-op). After IDLE is emitted we still wake the Java mailbox every idleTimeout for no useful status change → empty next() churn.

Suggestion: skip scheduleIdleTimer when already idle (re-arm only after ACTIVE / onRecord).

---

[Suggestion] Calling close() from the destructor

~WatermarkAssigner() {
  WatermarkAssigner::close();
}

StatefulTask::finish() usually already called close(). Prefer a dedicated shutdownScheduler() used by both close() and the destructor, and keep business close() out of ~T(). Same pattern in ~WatermarkSource().

[KevinyhZou]

[Nit] Stale comments

scheduleIdleTimer comment still refers to idleCheckRequested_, which was removed in commit 68e45f50. Please update to match the current design (timerPending_ is only a duplicate-registration guard; idle check always runs in checkWatermarkStatus).

Related: the scheduler_ comment (“Lazily created on the first record arrival”) is also inaccurate — creation happens on the first scheduleIdleTimer from checkWatermarkStatus.

[KevinyhZou]

[Blocker 2 / duplication] Same now + idleTimeout scheduling bug here

WatermarkSource copies the WatermarkAssigner timer pattern (fireAt = now + idleTimeout). Please fix both sites, and strongly consider extracting shared idle-timer management (IdleTimerManager / helper) so Assigner and Source cannot drift again.

Also note: after already IDLE, scheduleIdleTimer is still always called from checkWatermarkStatus (same as Blocker 3).

Minor: empty batches early-return before onRecord — confirm whether that matches the intended Flink semantics (any element vs non-empty data).

[KevinyhZou]

[Suggestion] Missing integration coverage

Unit tests for WatermarkIdleTracker look solid (including the repeated-baseline case). Please also add higher-level tests covering:

1. Blocked source + idle → same StatefulTask::next() returns WatermarkStatus (Blocker 1)
2. Timer fire time ≈ lastRecord + idleTimeout (Blocker 2)
3. Already-idle does not keep re-arming / does not re-emit IDLE (Blocker 3)
4. ACTIVE reactivation then second IDLE transition

Without these, the critical races in StatefulTask / Assigner / Source stay unguarded.