[server] Roll expired active segments for remote retention

[fxbing]

Purpose

Linked issue: close #3830

Local TTL cleanup currently scans only inactive log segments. When a table becomes idle after its active segment exceeds table.log.ttl, that segment remains active and cannot be uploaded to remote storage or removed by subsequent local retention.

This change rolls a non-empty expired active segment during periodic local retention once the high watermark reaches the log end offset. The rolled segment is not deleted in the same cleanup pass; it can first be uploaded and then removed by a later cleanup.

Leader/follower segment boundary alignment during timestamp-based roll is intentionally out of scope.

Brief change log

• Cache whether remote log is enabled from remote.log.task-interval-duration.
• Extend the contiguous TTL scan to inspect the active segment while preserving the existing remoteLogEndOffset and minRetainOffset boundaries for inactive segments.
• Roll the active segment only when remote log is enabled, the segment is non-empty and expired, and highWatermark >= logEndOffset.
• Do not require remoteLogEndOffset to cover the active segment before rolling, because rolling is required before the segment can be uploaded.
• Use the existing LogTablet.roll(Optional.empty()) path and narrow its exception declaration to IOException.
• Extend TieredLocalSegmentTTLTest for partitioned and non-partitioned tables, covering:
  • upload and cleanup lifecycle;
  • high-watermark gating;
  • remote offset boundaries;
  • disabled remote log;
  • disabled table TTL;
  • empty active segments.

Tests

• ./mvnw -pl fluss-server -DskipITs -Dcheckstyle.skip -Drat.skip -Dspotless.check.skip=true -Dtest=TieredLocalSegmentTTLTest test
  • 10 tests passed.
• ./mvnw -pl fluss-server spotless:check

API and Format

No public API, RPC protocol, configuration, or storage format changes.

Documentation

No documentation changes are required.

Generative AI disclosure

Yes. OpenAI Codex was used to assist with implementation and review.

[Copilot]

Pull request overview

This PR addresses #3830 by ensuring periodic TTL-based local retention can roll a non-empty, TTL-expired active log segment (once highWatermark >= logEndOffset), making the segment eligible for subsequent remote upload and later local cleanup.

Changes:

• Update LogTablet TTL retention logic to potentially roll an expired active segment (and narrow roll(...) to throws IOException).
• Extend tiered-storage TTL tests to cover active-segment roll behavior, high-watermark gating, remote end-offset boundaries, TTL disabled, and empty-active-segment behavior.
• Add a new non-tiered TTL test to validate behavior when remote log is disabled.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java	Rolls a TTL-expired active segment under safety conditions during TTL retention; narrows roll exception type.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java	Expands tiered TTL coverage to include expired active-segment rolling and related retention/upload boundaries.
fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java	Adds regression coverage for TTL cleanup behavior when remote log is disabled.

Suppressed comments (1)

fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:1383

• The PR description says rolling an expired active segment should only happen when remote log is enabled, but this condition currently depends only on segment size/TTL and high-watermark (no remote-log-enabled guard). If the intent really is remote-log-only, consider plumbing an explicit "remote log enabled" flag into LogTablet and gating this roll on it; otherwise, please align the PR description/scope with the implemented behavior.

        if (activeSegment.getSizeInBytes() > 0
                && isSegmentExpired(now, activeSegment, logTtlMs)
                && getHighWatermark() >= localLogEndOffset()) {
            roll(Optional.empty());
        }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.