refactor(inkless): stop retention and cleanup jobs during broker shutdown [KC-417]

[jeqo]

A broker restart could hang on an in-flight diskless retention or file-cleanup cycle. Both jobs ran on the shared KafkaScheduler, which drains its queue on shutdown without interrupting, waiting up to a day, and BrokerServer.shutdown awaits it early.

This moves both jobs onto a scheduler that ReplicaManager owns and makes each job stop at a safe point when closed. Both jobs are idempotent and re-derive unfinished work on the next cycle, so stopping loses no durable progress.

Changes

• Run RetentionEnforcer and FileCleaner on a dedicated 2-thread daemon scheduler while preserving their fixed-rate cadence.
• Stop new cycles, close both jobs, and wait up to 10s before using shutdownNow() as a backstop. The control plane and storage backend stay open during the grace period.
• Check the retention stop signal between partitions. Retention enforcement now issues one enforceRetention call per partition; the control plane already uses one transaction per partition, so this adds no transactions.
• Check the cleanup stop signal after the control-plane fetch and before object-storage deletion.
• Remove FileCleaner's in-task no-work and error sleeps. The periodic schedule now owns retry cadence.

Operator impact

• Shutdown waits at most 10s for these jobs instead of draining them for up to one day. If a job does not stop within the grace period, shutdown logs a warning, interrupts it, and proceeds.
• A restart that reaches a cooperative stop point does not increment retention or cleanup error metrics and does not log ERROR.
• FileCleaner retries an error on its next scheduled cycle rather than after an in-task backoff sleep.

Testing

• RetentionEnforcerTest: one control-plane call per enforceable partition, stop at a partition boundary, and run() after close() as a no-op.
• FileCleanerMockedTest: no on-thread sleep on no-work and error paths, deletion skipped when close() lands between fetch and delete, and run() after close() as a no-op.
• ReplicaManagerInklessTest: blocks a FileCleaner cycle mid-flight, starts shutdown on another thread, and asserts the shared state stays open until the job reaches its stop point. This exercises ReplicaManager owning the scheduler and the shutdown ordering itself, which the component tests above do not: they only cover each job's own stop flag, so they would pass unchanged even if shutdownInklessBackgroundJobs moved back after inklessSharedState.close().

[Copilot]

Pull request overview

Moves diskless retention and cleanup off Kafka’s shared scheduler to provide bounded broker shutdown.

Changes:

• Adds a dedicated stoppable background-job scheduler.
• Caps retention work per cycle and adds saturation metrics/configuration.
• Expands lifecycle, scheduling, and configuration tests.

Reviewed changes

Copilot reviewed 13 out of 13 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
core/src/main/scala/kafka/server/ReplicaManager.scala	Owns and shuts down the dedicated scheduler.
storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java	Adds the per-cycle partition cap.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlane.java	Clarifies retention API semantics.
storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java	Adds cooperative shutdown and removes sleeps.
storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java	Adds bounded, stoppable partition processing.
storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcementScheduler.java	Supports limited ready-partition polling.
storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcerMetrics.java	Adds cycle-saturation telemetry.
storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java	Tests the new configuration.
storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java	Tests cleaner shutdown behavior.
storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcerTest.java	Tests caps and cooperative stopping.
storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcementSchedulerTest.java	Tests limited polling and preserved lag.
docs/inkless/configs.rst	Documents the new configuration.
docs/inkless/metrics.rst	Documents the saturation metric.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

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

Suppressed comments (1)

storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java:53

• This lifecycle guarantee contradicts RetentionEnforcementScheduler.java:115-118, where getReadyPartitions immediately reschedules every returned partition. If close() stops this loop after the first request, the remaining fetched partitions don't keep their past-due queue slots. Describe the restart-based recovery instead, or requeue the unprocessed entries.

 * any later {@code run()} is a no-op. Enforcement is idempotent and per-partition independent, so an
 * abandoned cycle loses no durable progress: the partitions it did not reach keep their queue slot and are
 * enforced on a later cycle (here or on another broker).

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Suppressed comments (3)

storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java:53

• getReadyPartitions reschedules every returned partition before this loop starts, so partitions skipped after close() do not keep their queue slot; they move to a new enforcement time. Update this lifecycle contract to state that the work remains recoverable and is re-derived after restart rather than promising that the original slot remains due.

 * any later {@code run()} is a no-op. Enforcement is idempotent and per-partition independent, so an
 * abandoned cycle loses no durable progress: the partitions it did not reach keep their queue slot and are
 * enforced on a later cycle (here or on another broker).

storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java:265

• The dedicated scheduler uses a 500 ms fixed delay after each run, so a sustained backlog is capped at maxPartitionsPerCycle / (cycle duration + 500 ms). Leaving excess partitions overdue avoids waiting a full enforcement interval, but it does not guarantee that throughput is unchanged from an unbounded cycle. Reword this claim and regenerate docs/inkless/configs.rst.

        + "cycle; partitions left over stay due and are picked up by the next cycle (500ms later) rather than "
        + "waiting a full enforcement interval, so throughput is not lost. Watch "

storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java:148

• A sustained backlog reaches this branch after every cycle, and the scheduler can run it every 500 ms. This emits up to two INFO records per second per broker for a condition already exposed by RetentionEnforcementCycleSaturatedRate, creating avoidable operational log volume. Rate-limit this message or lower it to DEBUG.

            LOGGER.info("Retention enforcement cycle saturated at {} partitions; overdue partitions may remain queued",
                maxPartitionsPerCycle);

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (1)

core/src/main/scala/kafka/server/ReplicaManager.scala:3429

• The broker-level regression remains untested: the component tests cover each job's stop flag, but they don't exercise ReplicaManager owning the executor or prove that shutdown waits for an in-flight job before closing inklessSharedState. A later reorder of this call can restore the original shutdown failure while all added tests still pass. Add a ReplicaManagerInklessTest that blocks a cleaner/enforcer cycle, starts shutdown, and verifies shared state stays open until the job reaches its cooperative stop point.

    shutdownInklessBackgroundJobs()

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (1)

core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala:7701

• TIMED_WAITING doesn't identify bgScheduler.awaitTermination; any earlier shutdown step that uses a timed park can satisfy this predicate. The test would then release the job before shutdown reaches the ordering under test, so moving sharedState.close() ahead of the background-job join could pass. Wait for a stack containing both shutdownInklessBackgroundJobs and awaitTermination, or expose a latch at that join point.

      // Poll for the shutdown thread parking in bgScheduler.awaitTermination rather than sleeping a
      // fixed duration: this is the exact moment shutdownInklessBackgroundJobs() is waiting on the
      // blocked job, which is what this test needs to observe.
      waitUntilTrue(() => shutdownThread.getState == Thread.State.TIMED_WAITING,
        "shutdown thread never parked waiting for the in-flight background job")

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.