feat(inkless): Introduce CREATE_TOPICS config interceptor framework and DisklessForceCreateTopicInterceptor

[tvainika]

Introduce CREATE_TOPICS config interceptor framework and DisklessForceCreateTopicInterceptor

Summary

Introduce a generic interceptor framework for mutating topic configs during CREATE_TOPICS API requests, then use it to add a new diskless force interceptor.

Motivation

ClassicTopicRemoteStorageForcePolicy was implemented to automatically set remote.storage.enable=true on topic creation. For Diskless-enabled-by-default, we need a similar mechanism that forces diskless.enable=true. Rather than adding another one-off policy, this PR refactors the approach into a generic, chainable interceptor pattern.

Changes

Commit 1: Interceptor framework

• CreateTopicConfigInterceptor — interface with intercept() methods returning boolean (true = matched)
• CreateTopicConfigInterceptors — container that chains interceptors with first-match-wins semantics
• Refactored ClassicTopicRemoteStorageForcePolicy → ClassicTopicRemoteStorageForceCreateTopicInterceptor
• Updated ReplicationControlManager to use the new abstraction

Commit 2: Diskless force interceptor

• DisklessForceCreateTopicInterceptor — forces diskless.enable=true for topics matching an allow-list of regexes
• Excludes system topics, compacted topics, and topics with __ prefix
• Chained with higher priority than the classic remote storage interceptor
• Both interceptors can be enabled simultaneously (users configure regexes to avoid overlap)

New configs

Config	Type	Default	Description
diskless.force.enable	boolean	false	Enable the diskless force interceptor
diskless.force.include.topic.regexes	list	[]	Allow-list of regexes; matching topics get diskless.enable=true

Example configuration

# Enable both interceptors
classic.remote.storage.force.enable=true
classic.remote.storage.force.exclude.topic.regexes=diskless-.*

diskless.force.enable=true
diskless.force.include.topic.regexes=diskless-.*

Testing

• Unit tests for DisklessForceCreateTopicInterceptor (regex matching, exclusions, __ prefix skip)
• Unit tests for CreateTopicConfigInterceptors (chaining, first-match-wins, empty chain)
• Existing ClassicTopicRemoteStorageForceCreateTopicInterceptorTest continues to pass

[Copilot]

Pull request overview

This PR introduces a chainable interceptor framework for mutating topic configs during CREATE_TOPICS requests, and uses it to implement a new diskless “force enable” behavior (regex allow-list), while refactoring the existing classic remote-storage forcing logic into the same framework.

Changes:

• Added CreateTopicConfigInterceptor + CreateTopicConfigInterceptors (first-match-wins chain) and integrated it into ReplicationControlManager’s topic-creation flow.
• Refactored classic remote storage forcing into ClassicTopicRemoteStorageForceCreateTopicInterceptor.
• Added DisklessForceCreateTopicInterceptor with new server configs (diskless.force.enable, diskless.force.include.topic.regexes) plus unit tests.

Reviewed changes

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

Show a summary per file

File	Description
server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java	Adds new controller/server configs for enabling diskless forcing and defining the allow-list regexes.
core/src/main/scala/kafka/server/KafkaConfig.scala	Wires new diskless force configs into KafkaConfig.
core/src/main/scala/kafka/server/ControllerServer.scala	Passes diskless force settings into the controller builder.
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java	Threads diskless force config through QuorumController → ReplicationControlManager.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java	Replaces the old policy call-sites with interceptor-chain application during create-topics processing/validation/response building.
metadata/src/main/java/org/apache/kafka/controller/CreateTopicConfigInterceptor.java	Introduces the interceptor interface used for CREATE_TOPICS config mutation.
metadata/src/main/java/org/apache/kafka/controller/CreateTopicConfigInterceptors.java	Implements chaining + ordering (diskless first, then remote storage) with first-match-wins semantics.
metadata/src/main/java/org/apache/kafka/controller/DisklessForceCreateTopicInterceptor.java	New interceptor that forces diskless.enable=true for allow-listed topics with exclusions.
metadata/src/main/java/org/apache/kafka/controller/ClassicTopicRemoteStorageForceCreateTopicInterceptor.java	Refactors classic remote storage forcing into the new interceptor model.
metadata/src/test/java/org/apache/kafka/controller/CreateTopicConfigInterceptorsTest.java	Adds unit tests for chaining behavior and priority.
metadata/src/test/java/org/apache/kafka/controller/DisklessForceCreateTopicInterceptorTest.java	Adds unit tests for diskless forcing, exclusions, and __ prefix behavior.
metadata/src/test/java/org/apache/kafka/controller/ClassicTopicRemoteStorageForceCreateTopicInterceptorTest.java	Updates tests to reflect the refactor from policy to interceptor.

---

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

[Copilot]

Pull request overview

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