feat(inkless:control_plane): add dynamic diskless.control.plane.availability config

[gqmelo]

This new config is intended to allow a management plane report the state of the control plane coordinator service.

The config should be dynamicaly updated whenever the service becomes unavailable or available.

With that information, we then check for control plane availability before any control plane method, fetch requests and open/close the connection pool when the state changes.

KC-419

[Copilot]

Pull request overview

Adds a new dynamic cluster-wide broker config (diskless.control.plane.availability) that allows an external management plane to signal control plane availability, and uses that signal to proactively gate diskless operations (control-plane calls, produce/fetch hot paths, and periodic background tasks) to reduce timeouts and make failures immediately retriable.

Changes:

• Introduces ControlPlaneAvailability + AvailabilityGatedControlPlane, and wires them into broker/controller startup and dynamic reconfiguration.
• Gates diskless produce/fetch and skips periodic diskless background jobs when the control plane is reported unavailable.
• Adds metrics and documentation for control plane availability/gated-call counts, plus unit/integration tests for gating and dynamic updates.

Reviewed changes

Copilot reviewed 23 out of 23 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
storage/inkless/src/test/java/io/aiven/inkless/produce/AppendHandlerTest.java	Updates produce handler tests for availability gating and new constructor signature.
storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerIntegrationTest.java	Passes ControlPlaneAvailability into SharedState initialization for integration coverage.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/ControlPlaneAvailabilityTest.java	New unit tests for availability state handling, config parsing, and listener behavior.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/AvailabilityGatedControlPlaneTest.java	New unit tests for gating semantics and delegate lifecycle behavior.
storage/inkless/src/test/java/io/aiven/inkless/consume/FetchHandlerTest.java	Updates fetch handler tests for availability gating and retriable error behavior.
storage/inkless/src/test/java/io/aiven/inkless/common/SharedStateTest.java	Updates SharedState.initialize calls to include control plane availability.
storage/inkless/src/main/java/io/aiven/inkless/produce/AppendHandler.java	Gates diskless produce when control plane is reported unavailable.
storage/inkless/src/main/java/io/aiven/inkless/doc/MetricsDocs.java	Adds generated docs output for control plane availability metrics.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlaneAvailabilityMetrics.java	New Yammer/JMX metrics for reported availability and gated-call count.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlaneAvailability.java	New availability state holder with dynamic updates, metrics, and listener hook.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/AvailabilityGatedControlPlane.java	New wrapper that rejects calls while unavailable and lazily manages delegate lifecycle.
storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java	Gates diskless fetch when control plane is reported unavailable.
storage/inkless/src/main/java/io/aiven/inkless/common/SharedState.java	Threads ControlPlaneAvailability through shared state so handlers can gate consistently.
server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java	Defines new config key, allowed values, default, and docs; marks it dynamically updatable.
docs/inkless/metrics.rst	Documents the new control plane availability metrics in Inkless docs.
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala	Adds coverage for runIfControlPlaneAvailable gating helper.
core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala	Adds test ensuring dynamic updates propagate into ControlPlaneAvailability.
core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala	Adds tests for dynamic-config startup catch-up and config validation/update mode.
core/src/main/scala/kafka/server/SharedServer.scala	Creates and wires availability + availability-gated control plane during SharedServer start.
core/src/main/scala/kafka/server/ReplicaManager.scala	Skips periodic diskless tasks when gated; passes availability to fetch handler.
core/src/main/scala/kafka/server/KafkaConfig.scala	Exposes disklessControlPlaneAvailability on KafkaConfig.
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala	Registers dynamic reconfigurable to update availability; adds startup catch-up logic.
core/src/main/scala/kafka/server/BrokerServer.scala	Passes availability into SharedState.initialize.

Suppressed comments (1)

storage/inkless/src/test/java/io/aiven/inkless/produce/AppendHandlerTest.java:201

• This test creates an AppendHandler but does not close it, and it doesn't wait for the returned future. Closing via try-with-resources and awaiting the future makes the test more robust and avoids leaking any resources managed by AppendHandler.

    @Test
    void availableAppendDelegatesToWriter() {
        final Writer writer = Mockito.mock(Writer.class);
        Mockito.when(writer.write(Mockito.any(), Mockito.any(), Mockito.any()))
            .thenReturn(CompletableFuture.completedFuture(Map.of()));
        final AppendHandler appendHandler = new AppendHandler(
            writer, topic -> new LogConfig(Map.of()), AVAILABLE);

        appendHandler.handle(Map.of(T0P0, MemoryRecords.EMPTY), RequestLocal.noCaching());

        Mockito.verify(writer).write(Mockito.any(), Mockito.any(), Mockito.any());
    }

---

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