KAFKA-19082: [1/4] Add client config for enable2PC and overloaded initProducerId (KIP-939) #13
Conversation
…-Side-2PC-Changes-pt1
|
Warning Rate limit exceeded@visz11 has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 27 minutes and 26 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (9)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
|
/refacto-test |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Solid Implementation - Important Configuration Validation Needed
|
|
/refacto-test |
|
PR already reviewed at the latest commit: 4a08713. |
|
/refacto-test |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Transaction Management Implementation👍 Well Done
📌 Files Processed
📝 Additional Comments
|
![refacto-test[bot]](https://avatars.githubusercontent.com/u/196302493?s=60&v=4){kind=small}
| public void initTransactions(boolean keepPreparedTxn) { | ||
| throwIfNoTransactionManager(); | ||
| throwIfProducerClosed(); | ||
| long now = time.nanoseconds(); | ||
| TransactionalRequestResult result = transactionManager.initializeTransactions(); | ||
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); |
There was a problem hiding this comment.
Missing Validation Check
The new initTransactions method lacks validation for 2PC compatibility when keepPreparedTxn=true. Without validation, using keepPreparedTxn without enable2PC could cause transaction recovery failures.
| public void initTransactions(boolean keepPreparedTxn) { | |
| throwIfNoTransactionManager(); | |
| throwIfProducerClosed(); | |
| long now = time.nanoseconds(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); | |
| public void initTransactions(boolean keepPreparedTxn) { | |
| throwIfNoTransactionManager(); | |
| throwIfProducerClosed(); | |
| if (keepPreparedTxn && !transactionManager.isEnable2PC()) { | |
| throw new IllegalStateException("Cannot use keepPreparedTxn=true without enabling two-phase commit"); | |
| } | |
| long now = time.nanoseconds(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Preconditions
| synchronized TransactionalRequestResult initializeTransactions(ProducerIdAndEpoch producerIdAndEpoch) { | ||
| return initializeTransactions(producerIdAndEpoch, false); | ||
| } | ||
|
|
||
| public synchronized TransactionalRequestResult initializeTransactions(boolean keepPreparedTxn) { | ||
| return initializeTransactions(ProducerIdAndEpoch.NONE, keepPreparedTxn); | ||
| } | ||
|
|
There was a problem hiding this comment.
Boolean Parameter Clarity
Boolean parameters reduce readability and create implicit dependencies. Consider using enums or constants for the keepPreparedTxn parameter to improve code clarity and maintainability.
Standards
- Clean-Code-Boolean-Parameters
- Clean-Code-Meaningful-Names
| InitProducerIdRequestData requestData = new InitProducerIdRequestData() | ||
| .setTransactionalId(transactionalId) | ||
| .setTransactionTimeoutMs(transactionTimeoutMs) | ||
| .setProducerId(producerIdAndEpoch.producerId) | ||
| .setProducerEpoch(producerIdAndEpoch.epoch); | ||
| .setProducerEpoch(producerIdAndEpoch.epoch) | ||
| .setEnable2Pc(enable2PC) | ||
| .setKeepPreparedTxn(keepPreparedTxn); |
There was a problem hiding this comment.
InitProducerId Request Optimization
The InitProducerId request now includes two additional flags that increase request size. For high-throughput systems with frequent producer initialization, consider batching producer initializations or implementing a connection pooling strategy to amortize the cost of these larger requests.
Standards
- ISO-IEC-25010-Performance-Resource-Utilization
- Netflix-Hot-Path-Optimization
|
/refacto-test-arvi |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Transaction Management Enhancements👍 Well Done
📌 Files Processed
📝 Additional Comments
|
| boolean enable2PC = this.getBoolean(TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG); | ||
| boolean userConfiguredTransactionTimeout = originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG); | ||
| if (enable2PC && userConfiguredTransactionTimeout) { | ||
| throw new ConfigException( | ||
| "Cannot set " + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG + | ||
| " when " + ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG + | ||
| " is set to true. Transactions will not expire with two-phase commit enabled." | ||
| ); | ||
| } |
There was a problem hiding this comment.
Configuration Validation Gap
The validation prevents setting transaction.timeout.ms with 2PC enabled, but doesn't handle the default transaction timeout. The default timeout will still be used internally even when 2PC is enabled, potentially causing inconsistent behavior.
| boolean enable2PC = this.getBoolean(TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG); | |
| boolean userConfiguredTransactionTimeout = originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG); | |
| if (enable2PC && userConfiguredTransactionTimeout) { | |
| throw new ConfigException( | |
| "Cannot set " + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG + | |
| " when " + ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG + | |
| " is set to true. Transactions will not expire with two-phase commit enabled." | |
| ); | |
| } | |
| boolean enable2PC = this.getBoolean(TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG); | |
| boolean userConfiguredTransactionTimeout = originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG); | |
| if (enable2PC && userConfiguredTransactionTimeout) { | |
| throw new ConfigException( | |
| "Cannot set " + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG + | |
| " when " + ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG + | |
| " is set to true. Transactions will not expire with two-phase commit enabled." | |
| ); | |
| } | |
| // Override default transaction timeout when 2PC is enabled to ensure consistency | |
| if (enable2PC) { | |
| this.values().put(TRANSACTION_TIMEOUT_CONFIG, Integer.MAX_VALUE); | |
| } |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Precondition
| public void initTransactions() { | ||
| initTransactions(false); | ||
| } |
There was a problem hiding this comment.
Missing Default Handling
The default implementation of initTransactions() hardcodes 'false' for keepPreparedTxn. This prevents configuration-based defaults and forces explicit method calls for 2PC functionality, potentially causing inconsistent transaction behavior.
Standards
- ISO-IEC-25010-Reliability-Maturity
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Interface-Contract
| if (request instanceof InitProducerIdRequest) { | ||
| InitProducerIdRequest initRequest = (InitProducerIdRequest) request; | ||
| requestFlags[0] = initRequest.data().keepPreparedTxn(); |
There was a problem hiding this comment.
Null Safety Check
The test code accesses initRequest.data() without null checking. If data() returns null, this would cause a NullPointerException, making tests unreliable and potentially masking real issues.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- SRE-Error-Handling
|
/refacto-test-arvi |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Transaction Management Two-Phase Commit Support👍 Well Done
📌 Files Processed
📝 Additional Comments
|
| public void initTransactions(boolean keepPreparedTxn) { | ||
| throwIfNoTransactionManager(); | ||
| throwIfProducerClosed(); | ||
| long now = time.nanoseconds(); | ||
| TransactionalRequestResult result = transactionManager.initializeTransactions(); | ||
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); |
There was a problem hiding this comment.
Missing Transaction Validation
The new initTransactions(boolean) method doesn't validate that keepPreparedTxn=true is only used with enable2PC=true. Using keepPreparedTxn without 2PC enabled could cause transaction state inconsistencies since the broker won't be configured to handle prepared transactions properly.
| public void initTransactions(boolean keepPreparedTxn) { | |
| throwIfNoTransactionManager(); | |
| throwIfProducerClosed(); | |
| long now = time.nanoseconds(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); | |
| public void initTransactions(boolean keepPreparedTxn) { | |
| throwIfNoTransactionManager(); | |
| throwIfProducerClosed(); | |
| // Validate that keepPreparedTxn is only used with enable2PC | |
| if (keepPreparedTxn && !transactionManager.isEnable2PC()) { | |
| throw new IllegalStateException("Cannot use keepPreparedTxn=true without enabling two-phase commit. " + | |
| "Set " + ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG + "=true in the producer config."); | |
| } | |
| long now = time.nanoseconds(); | |
| TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn); |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Precondition-Validation
- SRE-Error-Prevention
| boolean userConfiguredTransactionTimeout = originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG); | ||
| if (enable2PC && userConfiguredTransactionTimeout) { | ||
| throw new ConfigException( |
There was a problem hiding this comment.
Configuration Validation Gap
Configuration validation only checks if transaction timeout is explicitly set, but doesn't validate when enable2PC=true with default transaction timeout. This creates inconsistent validation as the default timeout would still apply but isn't being rejected, potentially causing reliability issues in 2PC transactions.
| boolean userConfiguredTransactionTimeout = originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG); | |
| if (enable2PC && userConfiguredTransactionTimeout) { | |
| throw new ConfigException( | |
| boolean enable2PC = this.getBoolean(TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG); | |
| if (enable2PC) { | |
| // When 2PC is enabled, transaction timeout should not be used at all | |
| // as transactions are managed by an external coordinator | |
| if (originalConfigs.containsKey(TRANSACTION_TIMEOUT_CONFIG)) { | |
| throw new ConfigException( | |
| "Cannot set " + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG + | |
| " when " + ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG + | |
| " is set to true. Transactions will not expire with two-phase commit enabled." | |
| ); | |
| } | |
| // Force transaction timeout to a very large value when 2PC is enabled | |
| // to effectively disable timeout-based transaction expiration | |
| this.values().put(TRANSACTION_TIMEOUT_CONFIG, Integer.MAX_VALUE); | |
| } |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Precondition-Validation
- SRE-Configuration-Consistency
| /** | ||
| * Initialize the transactional state for this producer, similar to {@link #initTransactions()} but | ||
| * with additional capabilities to keep a previously prepared transaction. | ||
| * Must be called before any send operations that require a {@code transactionalId}. | ||
| * <p> | ||
| * Unlike the standard {@link #initTransactions()}, when {@code keepPreparedTxn} is set to | ||
| * {@code true}, the producer does <em>not</em> automatically abort existing transactions. | ||
| * Instead, it enters a recovery mode allowing only finalization of those previously prepared transactions. |
There was a problem hiding this comment.
Incomplete Documentation
Documentation doesn't clearly state that keepPreparedTxn=true requires enable2PC=true configuration. Missing this critical dependency information could lead to misuse of the API and transaction reliability issues when developers use keepPreparedTxn without enabling 2PC.
Standards
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- ISO-IEC-25010-Reliability-Maturity
- DbC-Interface-Documentation
- SRE-Error-Prevention
|
/refacto-test-arvi |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Transaction Management Implementation👍 Well Done
📌 Files Processed
📝 Additional Comments
|
| final ApiVersions apiVersions, | ||
| final boolean enable2PC) { |
There was a problem hiding this comment.
Transaction Timeout Validation
The validation prevents setting transaction.timeout.ms when 2PC is enabled, but doesn't handle the case where the default transaction.timeout.ms is still used internally. The TransactionManager constructor still receives the transactionTimeoutMs parameter without checking if 2PC is enabled.
| final ApiVersions apiVersions, | |
| final boolean enable2PC) { | |
| final ApiVersions apiVersions, | |
| final boolean enable2PC) { | |
| if (enable2PC && transactionTimeoutMs != Integer.MAX_VALUE) { | |
| log.info("Using MAX_VALUE for transaction timeout with 2PC enabled instead of {}", transactionTimeoutMs); | |
| } |
Standards
- ISO-IEC-25010-Reliability-Maturity
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Precondition
| producer.initTransactions(keepPreparedTxn); | ||
|
|
||
| // Verify request flags match expected values | ||
| assertEquals(keepPreparedTxn, requestFlags[0], | ||
| "keepPreparedTxn flag should match input parameter"); | ||
| assertEquals(enable2PC, requestFlags[1], | ||
| "enable2Pc flag should match producer configuration"); |
There was a problem hiding this comment.
Null Safety Check
The test accesses requestFlags array without checking if the InitProducerIdRequest was actually processed. If the request handler wasn't called, requestFlags would contain default values (false), potentially causing false test failures.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- SRE-Test-Reliability
3 participants
This is part of the client side changes required to enable 2PC for
KIP-939
Producer Config:
transaction.two.phase.commit.enable The default would be ‘false’. If
set to ‘true’, the broker is informed that the client is participating
in two phase commit protocol and transactions that this client starts
never expire.
Overloaded InitProducerId method
If the value is 'true' then the corresponding field is set in the
InitProducerIdRequest