[jira] [Comment Edited] (KAFKA-20029) Disallow partition count increase for __transaction_state.

["sanghyeok An (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20029?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18049037#comment-18049037 ] 

sanghyeok An edited comment on KAFKA-20029 at 1/5/26 12:07 AM:
---------------------------------------------------------------

Hi [~schofielaj] !

Thanks for the valuable feedback.

I completely agree with you. The `_{_}consumer_offsets` and `{_}_share_group_state` topics rely on similar hashing mechanisms, 
so they carry similar risks regarding partition expansion. 
I will update the scope of this issue to cover all internal topics and aim for a unified solution.

Regarding the implementation, I suggested two options in the description:

*1. Server-side enforcement (Controller/AdminManager):*

Rejecting *CreatePartitions* requests for internal topics by default. I also thought about introducing a safety config (e.g., {*}internal.topic.modification.enable{*}) to controller side for advanced users who might need to override this in emergencies.
    
*2. CLI Warning*
A softer guardrail in *kafka-topics.sh* (though this wouldn't prevent API calls).
    

My initial thought is that *controller-side enforcement* could provide the strongest guardrail, 
but I may be missing constraints or compatibility considerations. 
What do you think would be the most appropriate approach here?

Also, given this would change behavior by rejecting a request that used to be accepted, 
do you think this should go through the *KIP,* or is it reasonable to handle it as a safety improvement/bug fix (and document it in the release notes)?

Thanks for your time! 


was (Author: JIRAUSER303328):
Hi [~schofielaj] !

Thanks for the valuable feedback.

I completely agree with you. The `__consumer_offsets` and `__share_group_state` topics rely on similar hashing mechanisms, 
so they carry similar risks regarding partition expansion. 
I will update the scope of this issue to cover all internal topics and aim for a unified solution.

Regarding the implementation, I suggested two options in the description:

*1. Server-side enforcement (Controller/AdminManager):* 
- Rejecting *CreatePartitions* requests for internal topics by default. I also thought about introducing a safety config (e.g., {*}internal.topic.modification.enable{*}) to controller side for advanced users who might need to override this in emergencies.
    
*2. CLI Warning*
A softer guardrail in *kafka-topics.sh* (though this wouldn't prevent API calls).
    

My initial thought is that *controller-side enforcement* could provide the strongest guardrail, 
but I may be missing constraints or compatibility considerations. 
What do you think would be the most appropriate approach here?

Also, given this would change behavior by rejecting a request that used to be accepted, 
do you think this should go through the *KIP,* or is it reasonable to handle it as a safety improvement/bug fix (and document it in the release notes)?


Thanks for your time! 

> Disallow partition count increase for __transaction_state.
> ----------------------------------------------------------
>
>                 Key: KAFKA-20029
>                 URL: https://issues.apache.org/jira/browse/KAFKA-20029
>             Project: Kafka
>          Issue Type: Improvement
>            Reporter: sanghyeok An
>            Assignee: sanghyeok An
>            Priority: Minor
>
> The routing logic for the Transaction Coordinator relies heavily on the partition count of the internal topic {{{}__transaction_state{}}}. The mapping is determined by
> {code:java}
> Utils.abs(transactionId.hashCode()) % transactionTopicPartitionCount. {code}
> Consequently, changing the number of partitions for {{__transaction_state}} (e.g., expanding from 50 to 150) changes the mapping logic. While Kafka documentation advises against modifying internal topics, there is currently no hard guardrail preventing this via the Admin API.
> IMHO, Allowing this operation can lead to a split-brain scenario during a rolling upgrade or cluster expansion, resulting in orphan transactions, hanging transactions.
>  
> *Scenario & Analysis*
> Here is a breakdown of the race condition and inconsistency issues:
>  # *Expansion:* An admin expands {{__transaction_state}} partitions from 50 to 150. New partitions are created and leaders are elected.
>  # *Metadata Inconsistency:* While the metadata cache updates, the {{transactionTopicPartitionCount}} used by {{partitionsFor}} logic might not update atomically across all nodes.
>  # *Loading Failure:* A coordinator elected for a new partition might fail to load the transaction state in {{TransactionCoordinator#onElection}} due to a mismatch between the cached partition count and the actual partition count, throwing an exception.
>  # *Split-Brain View:*
>  ** *New Broker B* (starts up or receives update): Sees 150 partitions.
>  ** {*}Old Broker A{*}: Still operates with the logic based on 50 partitions.
>  # *Routing Divergence:*
>  ** A client sends {{FindCoordinator}} for Transaction ID "A".
>  ** *Broker B* calculates: {{hash("A") % 150}} -> Partition {*}130{*}.
>  ** *Broker A* calculates: {{hash("A") % 50}} -> Partition {*}27{*}.
>  # *non-deterministic behavior occur*
>  ** If producer send a FindCoordinator to *Broker B* for Transaction ID "A", it receives partition {*}130{*}. However, it send a request to {*}Broker A{*}, it receive partition {*}27{*}.
>  ** It means that producer will send {{InitProducerId}} to the leader of partition *130* or {*}27{*}. (anyway, the leader nodes of partition 27, 130 will be included in old brokers)
>  ** If TransactionCoordinator receive InitProducerID request for partition 130, it calls *partitionsFor(...)* to retrieve transaction state. but, TransactionCoordinator A is old broker and its *transactionTopicPartitionCount* is {*}50{*}. so result of re-calculating is {*}27{*}. It means that
>  *** Errors.COORDINATOR_LOAD_IN_PROGRESS can occur.
>  *** Transaction State will be write on unexpected __transaction_state partition. (for example, expect partition {*}130{*}, but partition *27.)*
>  **** it cause orphan transaction. for example, when the broker roll out, they try to restore transaction state. they expect that transaction of "A" will be in partition 130, but actually it will be in partition 11. So, data topic may encounter hanging transaction problem. 
>  ** In addition to, since the leaders for Partition 130 and Partition 27 are likely different, we now have two coordinators potentially managing the same Transaction ID without knowledge of each other.
>  
> *Impact*
> This non-deterministic behavior causes
>  * *Orphan/Hanging Transactions:* The producer interacts with a new coordinator that has no history of the transaction.
>  * *Availability Issues:* Clients may receive {{COORDINATOR_LOAD_IN_PROGRESS}} indefinitely if the coordinator fails to load the state due to count mismatch.
>  * Potential worst-case correctness risk (not asserted as guaranteed): depending on the timing, the transition could increase the risk of unexpected fencing/coordination behavior.
>  
>  
> To ensure cluster stability and data integrity, how about enforcing a guardrail against modifying the partition count of __transaction_state?
> *1. Enforce validation in Controller/AdminManager*
> It would be good to reject {{AlterPartitions}} requests for internal topics by default.
>  * Introduce a mechanism to check if the topic is internal during partition expansion.
>  * If {{topic.isInternal()}} is true, return an {{InvalidRequestException}} with a clear error message stating that internal topic partition counts cannot be changed dynamically.
>  * Introduce a safety configuration to controller side (e.g., {{{}internal.topic.modification.enable{}}}, default {{{}false{}}}) for advanced users who strictly need to override this behavior, although it is strongly discouraged.
> *2. CLI Warning*
> Add a warning message or require a *{{--force}}* flag in the *{{kafka-topics.sh}}* CLI when attempting to modify internal topics. Note that this is a soft guardrail and does not prevent programmatic changes via the Admin API. (For example, the go, rust, python CLI...)
>  
> Although the documentation explicitly advises against modifying the partition count of __transaction_state, the system currently permits it. This discrepancy creates a significant risk of critical human error. it would be good to enforce this safety constraint at the code level to ensure cluster stability.
> I would greatly appreciate kafka community's feedback on this issue.
> Thanks always. 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)