This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Current state: Under Discussion
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: KAFKA-20029
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Internal topics in Kafka, specifically __transaction_state, __consumer_offsets, and the recently introduced __share_group_state (KIP-932),
rely on a specific hashing mechanism to route keys (transaction IDs, group IDs) to partitions.
For example, the routing logic for the Transaction Coordinator maps a Transaction ID to a specific partition of __transaction_state using the following logic:
Utils.abs(transactionId.hashCode()) % transactionTopicPartitionCount |
This logic creates a strict dependency on the transactionTopicPartitionCount.
If the number of partitions changes (e.g., expanding from 50 to 150), the mapping logic changes immediately.
While the Kafka documentation advises against modifying these internal topics,
there is currently no hard guardrail preventing this operation via the Admin API CreatePartitions.
Allowing dynamic partition expansion for these internal topics leads to critical availability and correctness issues, including split-brain scenarios, orphan transactions, and hanging transactions.
Here is a breakdown of the race condition identified during an expansion scenario (e.g., 50 -> 150 partitions):
Metadata Inconsistency: During the expansion, the metadata cache updates are not atomic across all nodes.
Split-Brain View:
hash("A") % 50 -> Partition 27.hash("A") % 150 -> Partition 130.Routing Divergence: A producer sending FindCoordinator for Transaction ID "A" may be routed to Partition 130 or Partition 27 depending on which broker handles the request.
Orphan/Hanging Transactions:
If a client interacts with the new coordinator (Partition 130), it creates state there.
If the client later interacts with a broker using the old metadata (partition 50), it searches Partition 27.
As a result, the transaction state may be written to an unexpected partition, which can lead to orphan transactions that can no longer be correctly committed or aborted.
This issue is not limited to __transaction_state :
__consumer_offsets: Expansion causes existing consumer groups to lose track of their committed offsets because the groupId mapping shifts to a new partition.
__share_group_state: Similar routing logic applies to share groups, risking state inconsistency.
Currently, this is only prevented by documentation "advice," which is insufficient to prevent critical human errors or automated misconfigurations.
controller.allow.internal.topic.modification true, the controller allows partition expansion for internal topics. This is unsafe.INVALID_REQUEST : Returned for specific internal topics within a CreatePartitions response when expansion is attempted without the unsafe config enabled./kafka/core/src/main/scala/kafka/server/ControllerApis.scala → reject request
/kafka/core/src/main/scala/kafka/server/KafkaConfig.scala → Add configuration field for controller.allow.internal.topic.modification for ControllerApi.scala to access.
/kafka/raft/src/main/java/org/apache/kafka/raft/KRaftConfigs.java → Add controller.allow.internal.topic.modification.
We propose to enforce a validation check in the Controller API layer to reject CreatePartitions requests targeting internal topics,
while allowing other valid requests in the same batch to proceed.
We will introduce a new configuration to control this guardrail.
Config Name: controller.allow.internal.topic.modification
Definition: Defined in KRaftConfigs.java and propagated to KafkaConfig.scala.
Default: false
The validation logic will be implemented in kafka.server.ControllerApis to "fail fast" before the request reaches the Raft layer or Metadata cache.
In ControllerApis.handleCreatePartitions:
The controller will iterate through the list of topics in the CreatePartitionsRequest.
It will identify internal topics using Topic.isInternal(name).
If controller.allow.internal.topic.modification is false:
Requests for internal topics will be immediately marked with INVALID_REQUEST error in the response.
Valid (non-internal) topics in the same request will be passed to controller.createPartitions for processing.
The final response will merge the results, returning per-topic success or failure codes.
This approach ensures that an accidental inclusion of an internal topic in a bulk operation does not fail the entire batch (supporting partial failures),
aligning with standard Kafka Admin API behavior.
Behavioral Change (Operational Breaking Change):
This KIP introduces a breaking change in behavior. Previously, expanding internal topics was technically allowed (though unsafe). After this change, such requests will fail with INVALID_REQUEST.
This is a deliberate design choice to prioritize correctness and cluster stability over backward compatibility of an unsafe operation.
controller.allow.internal.topic.modification). Operators who strictly require this capability can explicitly enable it, ensuring that no critical workflows are permanently blocked.Client Compatibility:
Existing AdminClient implementations (Java, Go, Python, etc.) do not need to be updated. They will simply receive a standard INVALID_REQUEST error code, which is already handled by standard error handling routines. No binary incompatibility is introduced.
No features are being deprecated. This KIP strictly serves as a guardrail.
No Data Migration Required: This change applies only to metadata operations. Existing data in internal topics is unaffected.
For Administrators:
If an administrator must expand partitions (e.g., during a critical manual recovery scenario), they must effectively "opt-in" to the risk by dynamically updating the controller configuration controller.allow.internal.topic.modification to true.
Unit Tests (ControllerApisTest):
Batch Handling: Verify that a request containing [valid-topic, __transaction_state] returns NONE (success) for the valid topic and INVALID_REQUEST for the internal topic.
Config Override: Verify that expansion succeeds when controller.allow.internal.topic.modification is true.
Integration Tests:
Use AdminClient against a KRaft cluster fixture to verify end-to-end blocking behavior.
A theoretically more robust approach would be to adopt a partitioning strategy that minimizes key redistribution, such as Consistent Hashing. rejected reasons are:
Client Dependency: The partitioning logic (hashCode % partitionCount) is embedded in client libraries (FindCoordinator). Changing this algorithm requires updating the entire client ecosystem, creating a massive compatibility challenge.
Incomplete Solution: While Consistent Hashing reduces the scope of data redistribution, it does not eliminate it. Expanding the cluster would still require migrating a subset of keys (transaction states) to new partitions. As noted in the "Automatic Data Redistribution" section, implementing safe state migration for active, log-compacted internal topics is complex and risky, regardless of the hashing algorithm used.
Scope: The primary goal of this KIP is to prevent immediate availability risks in Kafka 4.0, not to redesign the core coordination and partitioning protocol.
We considered allowing the expansion and automatically redistributing data to new partitions. rejected reasons are:
Thundering Herd Risk (Metadata Storm): Since the routing logic is modular (hashCode % N), changing N re-maps almost all keys. This forces every active client to rediscover their coordinator simultaneously. This "Thundering Herd" would cause a massive spike in FindCoordinator requests, likely leading to request timeouts and a cluster-wide outage.
Complexity: Implementing a safe rebalancer for append-only log compacted topics involves complex concurrency controls that outweigh the benefits.
ReplicationControlManager (Deep Validation)We considered validating deep inside the metadata layer. rejected reason is:
ControllerApis) is more efficient and protects the critical path of the KRaft controller.We considered implementing the restriction solely in client-side tools (e.g., kafka-topics.sh) or the AdminClient SDK. rejected reasons are:
Lack of Universal Coverage: The Kafka ecosystem includes diverse client libraries (Java, C/C++, Go, Python, Rust, etc.). Enforcing this logic in every client implementation is impractical and prone to inconsistency.
Version Compatibility: Even if the Java AdminClient were updated to block this request, the cluster would remain vulnerable to older client versions that do not have this check.