[fxbing]

• No generative AI tools used
• Yes (please specify the tool below)

OpenAI Codex assisted with the implementation, tests, and PR preparation.

Purpose

Linked issue: close #4878

Writer expiration compared batch timestamps from the injected Clock with system time. Under a ManualClock, a fresh batch could be treated as expired, allowing sequence 1 to be accepted before a delayed sequence 0 and causing the earlier batch to be treated as committed on retry.

Brief change log

• Use the injected clock for online writer expiration and scheduled writer cleanup in LogTablet.
• Pass that clock through LogLoader for writer-state snapshot/replay rebuild. Retain system time for elapsed-duration logging.
• Add deterministic regressions for first-sequence validation, periodic cleanup, and snapshot reload.

Tests

• Three deterministic ManualClock regressions cover first-sequence validation, periodic expiration, and snapshot reload. They use an explicit 10-second expiration interval; the append test also verifies that persisted batch timestamps are zero.
• Final red-green verification: restoring system-time checks makes exactly these three tests fail (23 other tests pass); restoring the fix makes all 26 LogTabletTest tests pass.
• Before the final test-only enhancements, 55 tests passed across LogTabletTest, LogLoaderTest, and WriterStateManagerTest, and all four cases of FlinkTableSourceITCase#testLookupCustomShuffle (inherited by Flink22TableSourceITCase, executed in fluss-flink-2.2) passed on Java 11. Production code is unchanged since those runs.
• Checkstyle, Spotless, and license checks passed. The full project suite was not run. Changes are confined to the fluss-server log package and its tests; client and Flink production code is unchanged.

API and Format

No public API, protocol, or storage-format change.

Documentation

No documentation change.