You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jo...@apache.org on 2026/09/04 16:41:22 UTC
(kafka) branch 4.4 updated: KAFKA-19414: Revert 2PC public API changes for 4.4 (#23356)
This is an automated email from the ASF dual-hosted git repository.
jolshan pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new 31e698de7fd KAFKA-19414: Revert 2PC public API changes for 4.4 (#23356)
31e698de7fd is described below
commit 31e698de7fd1c6fd0bb46c727ca37c9069be28aa
Author: Ritika Reddy <98...@users.noreply.github.com>
AuthorDate: Fri Sep 4 09:41:15 2026 -0700
KAFKA-19414: Revert 2PC public API changes for 4.4 (#23356)
We are removing some of the previously added public APIs until KIP-939
is ready to use. Similar to the [4.3
PR](https://github.com/apache/kafka/pull/21902).
This is a cherry-pick of the 4.3 revert (`c41ff4de0e`) onto 4.4. Five of
the six files applied cleanly; the conflicts below were resolved
manually:
### `clients/.../producer/KafkaProducer.java`
1. **Import block** — 4.4 relocated `AppInfoParser`, `LogContext`, and
`ProducerIdAndEpoch` to `org.apache.kafka.common.utils.internals.*`.
Kept 4.4's import layout and dropped the now-unused
`internals.ProducerIdAndEpoch` import (the revert removed its only
usages).
2. **`prepareTransaction()` region** — this overlapped with a new 4.4
method, `awaitTopicMetadata()`. Kept `awaitTopicMetadata()` and removed
only the reverted `prepareTransaction()` method and its javadoc.
### `clients/.../producer/KafkaProducerTest.java`
The revert only removed the old
`org.apache.kafka.common.utils.ProducerIdAndEpoch` import path; in 4.4
that import had moved to `...utils.internals.ProducerIdAndEpoch`, so it
was left behind and is now unused after the revert. Removed it to keep
checkstyle happy.
The internal `TransactionManager` 2PC plumbing is untouched, matching
the scope of the 4.3 PR.
Reviewers: Justine Olshan <jo...@confluent.io>
---
.../kafka/clients/producer/KafkaProducer.java | 139 +---------
.../kafka/clients/producer/MockProducer.java | 37 +--
.../apache/kafka/clients/producer/Producer.java | 19 +-
.../producer/internals/TransactionManager.java | 4 +-
.../kafka/clients/producer/KafkaProducerTest.java | 298 ---------------------
.../producer/internals/TransactionManagerTest.java | 131 ---------
6 files changed, 14 insertions(+), 614 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 109306b6402..4f31161feb2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -54,13 +54,11 @@ import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidTopicException;
-import org.apache.kafka.common.errors.InvalidTxnStateException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.errors.TimeoutException;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -85,7 +83,6 @@ import org.apache.kafka.common.utils.Timer;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.AppInfoParser;
import org.apache.kafka.common.utils.internals.LogContext;
-import org.apache.kafka.common.utils.internals.ProducerIdAndEpoch;
import org.slf4j.Logger;
@@ -683,13 +680,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
}
/**
- * Initialize the transactional state for this producer, similar to {@link #initTransactions()} but
- * with additional capabilities to keep a previously prepared transaction.
- *
* Needs to be called before any other methods when the {@code transactional.id} is set in the configuration.
- *
- * When {@code keepPreparedTxn} is {@code false}, this behaves like the standard transactional
- * initialization where the method does the following:
+ * This method does the following:
* <ol>
* <li>Ensures any transactions initiated by previous instances of the producer with the same
* {@code transactional.id} are completed. If the previous instance had failed with a transaction in
@@ -698,39 +690,26 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* <li>Gets the internal producer id and epoch, used in all future transactional
* messages issued by the producer.</li>
* </ol>
- *
- * <p>
- * 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.
- * This behavior is especially crucial for 2PC scenarios, where transactions should remain intact
- * until the external transaction manager decides whether to commit or abort.
- * <p>
- *
- * @param keepPreparedTxn true to retain any in-flight prepared transactions (necessary for 2PC
- * recovery), false to abort existing transactions and behave like
- * the standard initTransactions.
- *
* Note that this method will raise {@link TimeoutException} if the transactional state cannot
* be initialized before expiration of {@code max.block.ms}. Additionally, it will raise {@link InterruptException}
* if interrupted. It is safe to retry in either case, but once the transactional state has been successfully
* initialized, this method should no longer be used.
*
- * @throws IllegalStateException if no {@code transactional.id} is configured
- * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the broker does not
- * support transactions (i.e. if its version is lower than 0.11.0.0)
- * @throws org.apache.kafka.common.errors.TransactionalIdAuthorizationException if the configured
- * {@code transactional.id} is unauthorized either for normal transaction writes or 2PC.
- * @throws KafkaException if the producer encounters a fatal error or any other unexpected error
+ * @throws IllegalStateException if no {@code transactional.id} has been configured
+ * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
+ * does not support transactions (i.e. if its version is lower than 0.11.0.0)
+ * @throws org.apache.kafka.common.errors.AuthorizationException error indicating that the configured
+ * transactional.id is not authorized, or the idempotent producer id is unavailable. See the exception for
+ * more details. User may retry this function call after fixing the permission.
+ * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
* @throws TimeoutException if the time taken for initialize the transaction has surpassed <code>max.block.ms</code>.
* @throws InterruptException if the thread is interrupted while blocked
*/
- public void initTransactions(boolean keepPreparedTxn) {
+ public void initTransactions() {
throwIfNoTransactionManager();
throwIfProducerClosed();
- throwIfInPreparedState();
long now = time.nanoseconds();
- TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn);
+ TransactionalRequestResult result = transactionManager.initializeTransactions(false);
sender.wakeup();
result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, INIT_TXN_TIMEOUT_MSG);
producerMetrics.recordInit(time.nanoseconds() - now);
@@ -755,7 +734,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
public void beginTransaction() throws ProducerFencedException {
throwIfNoTransactionManager();
throwIfProducerClosed();
- throwIfInPreparedState();
long now = time.nanoseconds();
transactionManager.beginTransaction();
producerMetrics.recordBeginTxn(time.nanoseconds() - now);
@@ -818,7 +796,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
throwIfInvalidGroupMetadata(groupMetadata);
throwIfNoTransactionManager();
throwIfProducerClosed();
- throwIfInPreparedState();
if (!offsets.isEmpty()) {
long start = time.nanoseconds();
@@ -858,49 +835,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
return TimeUnit.NANOSECONDS.toMillis(elapsedNanos);
}
- /**
- * Prepares the current transaction for a two-phase commit. This method will flush all pending messages
- * and transition the producer into a mode where only {@link #commitTransaction()}, {@link #abortTransaction()},
- * or completeTransaction(PreparedTxnState) may be called.
- * <p>
- * This method is used as part of a two-phase commit protocol:
- * <ol>
- * <li>Prepare the transaction by calling this method. This returns a {@link PreparedTxnState} if successful.</li>
- * <li>Make any external system changes that need to be atomic with this transaction.</li>
- * <li>Complete the transaction by calling {@link #commitTransaction()}, {@link #abortTransaction()} or
- * completeTransaction(PreparedTxnState).</li>
- * </ol>
- *
- * @return the prepared transaction state to use when completing the transaction
- *
- * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started yet.
- * @throws InvalidTxnStateException if the producer is not in a state where preparing
- * a transaction is possible or 2PC is not enabled.
- * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
- * @throws UnsupportedVersionException fatal error indicating the broker
- * does not support transactions (i.e. if its version is lower than 0.11.0.0)
- * @throws AuthorizationException fatal error indicating that the configured
- * transactional.id is not authorized. See the exception for more details
- * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
- * @throws TimeoutException if the time taken for preparing the transaction has surpassed <code>max.block.ms</code>
- * @throws InterruptException if the thread is interrupted while blocked
- */
- @Override
- public PreparedTxnState prepareTransaction() throws ProducerFencedException {
- throwIfNoTransactionManager();
- throwIfProducerClosed();
- throwIfInPreparedState();
- if (!transactionManager.is2PCEnabled()) {
- throw new InvalidTxnStateException("Cannot prepare a transaction when 2PC is not enabled");
- }
- long now = time.nanoseconds();
- flush();
- transactionManager.prepareTransaction();
- producerMetrics.recordPrepareTxn(time.nanoseconds() - now);
- ProducerIdAndEpoch producerIdAndEpoch = transactionManager.preparedTransactionState();
- return new PreparedTxnState(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
- }
-
/**
* Commits the ongoing transaction. This method will flush any unsent records before actually committing the transaction.
* <p>
@@ -978,41 +912,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
producerMetrics.recordAbortTxn(time.nanoseconds() - abortStart);
}
- /**
- * Completes a prepared transaction by comparing the provided prepared transaction state with the
- * current prepared state on the producer.
- * If they match, the transaction is committed; otherwise, it is aborted.
- *
- * @param preparedTxnState The prepared transaction state to compare against the current state
- * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started
- * @throws InvalidTxnStateException if the producer is not in prepared state
- * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
- * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
- * @throws TimeoutException if the time taken for completing the transaction has surpassed <code>max.block.ms</code>
- * @throws InterruptException if the thread is interrupted while blocked
- */
- @Override
- public void completeTransaction(PreparedTxnState preparedTxnState) throws ProducerFencedException {
- throwIfNoTransactionManager();
- throwIfProducerClosed();
-
- if (!transactionManager.isPrepared()) {
- throw new InvalidTxnStateException("Cannot complete transaction because no transaction has been prepared. " +
- "Call prepareTransaction() first, or make sure initTransaction(true) was called.");
- }
-
- // Get the current prepared transaction state
- ProducerIdAndEpoch currentProducerIdAndEpoch = transactionManager.preparedTransactionState();
- PreparedTxnState currentPreparedState = new PreparedTxnState(currentProducerIdAndEpoch.producerId, currentProducerIdAndEpoch.epoch);
-
- // Compare the prepared transaction state token and commit or abort accordingly
- if (currentPreparedState.equals(preparedTxnState)) {
- commitTransaction();
- } else {
- abortTransaction();
- }
- }
-
/**
* Asynchronously send a record to a topic. Equivalent to <code>send(record, null)</code>.
* See {@link #send(ProducerRecord, Callback)} for details.
@@ -1151,23 +1050,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
throw new IllegalStateException("Cannot perform operation after producer has been closed");
}
- /**
- * Throws an exception if the transaction is in a prepared state.
- * In a two-phase commit (2PC) flow, once a transaction enters the prepared state,
- * only commit, abort, or complete operations are allowed.
- *
- * @throws IllegalStateException if any other operation is attempted in the prepared state.
- */
- private void throwIfInPreparedState() {
- if (transactionManager != null &&
- transactionManager.isTransactional() &&
- transactionManager.isPrepared()
- ) {
- throw new IllegalStateException("Cannot perform operation while the transaction is in a prepared state. " +
- "Only commitTransaction(), abortTransaction(), or completeTransaction() are permitted.");
- }
- }
-
/**
* Implementation of asynchronously send a record to a topic.
*/
@@ -1179,7 +1061,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
try {
throwIfProducerClosed();
- throwIfInPreparedState();
// first make sure the metadata for the topic is available
long nowMs = time.milliseconds();
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
index 930cdc5d6b4..fc2fea014c5 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
@@ -78,7 +78,7 @@ public class MockProducer<K, V> implements Producer<K, V> {
private long commitCount = 0L;
private final List<KafkaMetric> addedMetrics = new ArrayList<>();
- /** Exception to throw when {@link #initTransactions(boolean)} is called */
+ /** Exception to throw when {@link #initTransactions()} is called */
public RuntimeException initTransactionException = null;
/** Exception to throw when {@link #beginTransaction()} is called */
public RuntimeException beginTransactionException = null;
@@ -153,7 +153,7 @@ public class MockProducer<K, V> implements Producer<K, V> {
}
@Override
- public void initTransactions(boolean keepPreparedTxn) {
+ public void initTransactions() {
verifyNotClosed();
verifyNotFenced();
if (this.transactionInitialized) {
@@ -211,18 +211,6 @@ public class MockProducer<K, V> implements Producer<K, V> {
this.sentOffsets = true;
}
- @Override
- public PreparedTxnState prepareTransaction() throws ProducerFencedException {
- verifyNotClosed();
- verifyNotFenced();
- verifyTransactionsInitialized();
- verifyTransactionInFlight();
-
- // Return a new PreparedTxnState with mock values for producerId and epoch
- // Using 1000L and (short)1 as arbitrary values for a valid PreparedTxnState
- return new PreparedTxnState(1000L, (short) 1);
- }
-
@Override
public void commitTransaction() throws ProducerFencedException {
verifyNotClosed();
@@ -268,27 +256,6 @@ public class MockProducer<K, V> implements Producer<K, V> {
this.transactionInFlight = false;
}
- @Override
- public void completeTransaction(PreparedTxnState preparedTxnState) throws ProducerFencedException {
- verifyNotClosed();
- verifyNotFenced();
- verifyTransactionsInitialized();
-
- if (!this.transactionInFlight) {
- throw new IllegalStateException("There is no prepared transaction to complete.");
- }
-
- // For testing purposes, we'll consider a prepared state with producerId=1000L and epoch=1 as valid
- // This should match what's returned in prepareTransaction()
- PreparedTxnState currentState = new PreparedTxnState(1000L, (short) 1);
-
- if (currentState.equals(preparedTxnState)) {
- commitTransaction();
- } else {
- abortTransaction();
- }
- }
-
private synchronized void verifyNotClosed() {
if (this.closed) {
throw new IllegalStateException("MockProducer is already closed.");
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java b/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java
index 68a26085583..c19cbedc1a6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java
@@ -44,14 +44,7 @@ public interface Producer<K, V> extends Closeable {
/**
* See {@link KafkaProducer#initTransactions()}
*/
- default void initTransactions() {
- initTransactions(false);
- }
-
- /**
- * See {@link KafkaProducer#initTransactions(boolean)}
- */
- void initTransactions(boolean keepPreparedTxn);
+ void initTransactions();
/**
* See {@link KafkaProducer#beginTransaction()}
@@ -64,11 +57,6 @@ public interface Producer<K, V> extends Closeable {
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
ConsumerGroupMetadata groupMetadata) throws ProducerFencedException;
- /**
- * See {@link KafkaProducer#prepareTransaction()}
- */
- PreparedTxnState prepareTransaction() throws ProducerFencedException;
-
/**
* See {@link KafkaProducer#commitTransaction()}
*/
@@ -79,11 +67,6 @@ public interface Producer<K, V> extends Closeable {
*/
void abortTransaction() throws ProducerFencedException;
- /**
- * See {@link KafkaProducer#completeTransaction(PreparedTxnState)}
- */
- void completeTransaction(PreparedTxnState preparedTxnState) throws ProducerFencedException;
-
/**
* See {@link KafkaProducer#registerMetricForSubscription(KafkaMetric)}
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
index ebd238875f2..ef43c1c3fcd 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
@@ -344,9 +344,7 @@ public class TransactionManager {
.setTransactionalId(transactionalId)
.setTransactionTimeoutMs(transactionTimeoutMs)
.setProducerId(producerIdAndEpoch.producerId)
- .setProducerEpoch(producerIdAndEpoch.epoch)
- .setEnable2Pc(enable2PC)
- .setKeepPreparedTxn(keepPreparedTxn);
+ .setProducerEpoch(producerIdAndEpoch.epoch);
InitProducerIdHandler handler = new InitProducerIdHandler(new InitProducerIdRequest.Builder(requestData),
isEpochBump);
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index a47fbcf5846..e1de037bf4b 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -34,7 +34,6 @@ import org.apache.kafka.clients.producer.internals.ProducerMetadata;
import org.apache.kafka.clients.producer.internals.RecordAccumulator;
import org.apache.kafka.clients.producer.internals.Sender;
import org.apache.kafka.clients.producer.internals.TransactionManager;
-import org.apache.kafka.clients.producer.internals.TransactionalRequestResult;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
@@ -50,7 +49,6 @@ import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.errors.ClusterAuthorizationException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidTopicException;
-import org.apache.kafka.common.errors.InvalidTxnStateException;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
@@ -81,7 +79,6 @@ import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
import org.apache.kafka.common.requests.EndTxnResponse;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
-import org.apache.kafka.common.requests.InitProducerIdRequest;
import org.apache.kafka.common.requests.InitProducerIdResponse;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.MetadataResponse;
@@ -99,7 +96,6 @@ import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Timer;
import org.apache.kafka.common.utils.internals.LogContext;
-import org.apache.kafka.common.utils.internals.ProducerIdAndEpoch;
import org.apache.kafka.test.MockMetricsReporter;
import org.apache.kafka.test.MockPartitioner;
import org.apache.kafka.test.MockProducerInterceptor;
@@ -112,7 +108,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -168,7 +163,6 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.notNull;
import static org.mockito.Mockito.atMostOnce;
-import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
@@ -1390,298 +1384,6 @@ public class KafkaProducerTest {
}
}
- @ParameterizedTest
- @CsvSource({
- "true, false",
- "true, true",
- "false, true"
- })
- public void testInitTransactionsWithKeepPreparedTxnAndTwoPhaseCommit(boolean keepPreparedTxn, boolean enable2PC) {
- Map<String, Object> configs = new HashMap<>();
- configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-txn-id");
- configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
- configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
- if (enable2PC) {
- configs.put(ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG, true);
- }
-
- Time time = new MockTime(1);
- MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
- ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
- MockClient client = new MockClient(time, metadata);
- client.updateMetadata(initialUpdateResponse);
-
- // Capture flags from the InitProducerIdRequest
- boolean[] requestFlags = new boolean[2]; // [keepPreparedTxn, enable2Pc]
-
- client.prepareResponse(
- request -> request instanceof FindCoordinatorRequest &&
- ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
- FindCoordinatorResponse.prepareResponse(Errors.NONE, "test-txn-id", NODE));
-
- client.prepareResponse(
- request -> {
- if (request instanceof InitProducerIdRequest) {
- InitProducerIdRequest initRequest = (InitProducerIdRequest) request;
- requestFlags[0] = initRequest.data().keepPreparedTxn();
- requestFlags[1] = initRequest.data().enable2Pc();
- return true;
- }
- return false;
- },
- initProducerIdResponse(1L, (short) 5, Errors.NONE));
-
- try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
- new StringSerializer(), metadata, client, null, time)) {
- 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");
- }
- }
-
- @Test
- public void testPrepareTransactionSuccess() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- when(ctx.transactionManager.isTransactionV2Enabled()).thenReturn(true);
- when(ctx.transactionManager.is2PCEnabled()).thenReturn(true);
- when(ctx.sender.isRunning()).thenReturn(true);
-
- doNothing().when(ctx.transactionManager).prepareTransaction();
-
- long expectedProducerId = 12345L;
- short expectedEpoch = 5;
- ProducerIdAndEpoch expectedProducerIdAndEpoch = new ProducerIdAndEpoch(expectedProducerId, expectedEpoch);
- when(ctx.transactionManager.preparedTransactionState()).thenReturn(expectedProducerIdAndEpoch);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- PreparedTxnState returned = producer.prepareTransaction();
- assertEquals(expectedProducerId, returned.producerId());
- assertEquals(expectedEpoch, returned.epoch());
-
- verify(ctx.transactionManager).prepareTransaction();
- verify(ctx.accumulator).beginFlush();
- verify(ctx.accumulator).awaitFlushCompletion();
- }
- }
-
- @Test
- public void testSendNotAllowedInPreparedTransactionState() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- String topic = "foo";
- Cluster cluster = TestUtils.singletonCluster(topic, 1);
-
- when(ctx.sender.isRunning()).thenReturn(true);
- when(ctx.metadata.fetch()).thenReturn(cluster);
-
- // Mock transaction manager to simulate being in a prepared state
- when(ctx.transactionManager.isTransactional()).thenReturn(true);
- when(ctx.transactionManager.isPrepared()).thenReturn(true);
-
- // Create record to send
- long timestamp = ctx.time.milliseconds();
- ProducerRecord<String, String> record = new ProducerRecord<>(topic, 0, timestamp, "key", "value");
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- // Verify that sending a record throws IllegalStateException with the correct message
- IllegalStateException exception = assertThrows(
- IllegalStateException.class,
- () -> producer.send(record)
- );
-
- assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
-
- // Verify transactionManager methods were called
- verify(ctx.transactionManager).isTransactional();
- verify(ctx.transactionManager).isPrepared();
-
- // Verify that no message was actually sent (accumulator was not called)
- verify(ctx.accumulator, never()).append(
- eq(topic),
- anyInt(),
- anyLong(),
- any(),
- any(),
- any(),
- any(),
- anyLong(),
- anyLong(),
- any()
- );
- }
- }
-
- @SuppressWarnings("removal")
- @Test
- public void testSendOffsetsNotAllowedInPreparedTransactionState() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- String topic = "foo";
- Cluster cluster = TestUtils.singletonCluster(topic, 1);
-
- when(ctx.sender.isRunning()).thenReturn(true);
- when(ctx.metadata.fetch()).thenReturn(cluster);
-
- // Mock transaction manager to simulate being in a prepared state
- when(ctx.transactionManager.isTransactional()).thenReturn(true);
- when(ctx.transactionManager.isPrepared()).thenReturn(true);
-
- // Create consumer group metadata
- String groupId = "test-group";
- Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
- offsets.put(new TopicPartition(topic, 0), new OffsetAndMetadata(100L));
- ConsumerGroupMetadata groupMetadata = new ConsumerGroupMetadata(groupId);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- // Verify that sending offsets throws IllegalStateException with the correct message
- IllegalStateException exception = assertThrows(
- IllegalStateException.class,
- () -> producer.sendOffsetsToTransaction(offsets, groupMetadata)
- );
-
- assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
-
- // Verify transactionManager methods were called
- verify(ctx.transactionManager).isTransactional();
- verify(ctx.transactionManager).isPrepared();
-
- // Verify that no offsets were actually sent
- verify(ctx.transactionManager, never()).sendOffsetsToTransaction(
- eq(offsets),
- eq(groupMetadata)
- );
- }
- }
-
- @Test
- public void testBeginTransactionNotAllowedInPreparedTransactionState() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- when(ctx.sender.isRunning()).thenReturn(true);
-
- // Mock transaction manager to simulate being in a prepared state
- when(ctx.transactionManager.isTransactional()).thenReturn(true);
- when(ctx.transactionManager.isPrepared()).thenReturn(true);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- // Verify that calling beginTransaction throws IllegalStateException with the correct message
- IllegalStateException exception = assertThrows(
- IllegalStateException.class,
- producer::beginTransaction
- );
-
- assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
-
- // Verify transactionManager methods were called
- verify(ctx.transactionManager).isTransactional();
- verify(ctx.transactionManager).isPrepared();
- }
- }
-
- @Test
- public void testPrepareTransactionFailsWhen2PCDisabled() {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- // Disable 2PC
- when(ctx.transactionManager.isTransactionV2Enabled()).thenReturn(true);
- when(ctx.transactionManager.is2PCEnabled()).thenReturn(false);
- when(ctx.sender.isRunning()).thenReturn(true);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- assertThrows(
- InvalidTxnStateException.class,
- producer::prepareTransaction,
- "prepareTransaction() should fail if 2PC is disabled"
- );
- }
- }
-
- @Test
- public void testCompleteTransactionWithMatchingState() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- when(ctx.transactionManager.isPrepared()).thenReturn(true);
- when(ctx.sender.isRunning()).thenReturn(true);
-
- // Create prepared states with matching values
- long producerId = 12345L;
- short epoch = 5;
- PreparedTxnState inputState = new PreparedTxnState(producerId, epoch);
- ProducerIdAndEpoch currentProducerIdAndEpoch = new ProducerIdAndEpoch(producerId, epoch);
-
- // Set up the transaction manager to return the prepared state
- when(ctx.transactionManager.preparedTransactionState()).thenReturn(currentProducerIdAndEpoch);
-
- // Should trigger commit when states match
- TransactionalRequestResult commitResult = mock(TransactionalRequestResult.class);
- when(ctx.transactionManager.beginCommit()).thenReturn(commitResult);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- // Call completeTransaction with the matching state
- producer.completeTransaction(inputState);
-
- // Verify methods called in order
- verify(ctx.transactionManager).isPrepared();
- verify(ctx.transactionManager).preparedTransactionState();
- verify(ctx.transactionManager).beginCommit();
-
- // Verify abort was never called
- verify(ctx.transactionManager, never()).beginAbort();
-
- // Verify sender was woken up
- verify(ctx.sender).wakeup();
- }
- }
-
- @Test
- public void testCompleteTransactionWithNonMatchingState() throws Exception {
- StringSerializer serializer = new StringSerializer();
- KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
-
- when(ctx.transactionManager.isPrepared()).thenReturn(true);
- when(ctx.sender.isRunning()).thenReturn(true);
-
- // Create txn prepared states with different values
- long producerId = 12345L;
- short epoch = 5;
- PreparedTxnState inputState = new PreparedTxnState(producerId + 1, epoch);
- ProducerIdAndEpoch currentProducerIdAndEpoch = new ProducerIdAndEpoch(producerId, epoch);
-
- // Set up the transaction manager to return the prepared state
- when(ctx.transactionManager.preparedTransactionState()).thenReturn(currentProducerIdAndEpoch);
-
- // Should trigger abort when states don't match
- TransactionalRequestResult abortResult = mock(TransactionalRequestResult.class);
- when(ctx.transactionManager.beginAbort()).thenReturn(abortResult);
-
- try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
- // Call completeTransaction with the non-matching state
- producer.completeTransaction(inputState);
-
- // Verify methods called in order
- verify(ctx.transactionManager).isPrepared();
- verify(ctx.transactionManager).preparedTransactionState();
- verify(ctx.transactionManager).beginAbort();
-
- // Verify commit was never called
- verify(ctx.transactionManager, never()).beginCommit();
-
- // Verify sender was woken up
- verify(ctx.sender).wakeup();
- }
- }
-
@Test
public void testClusterAuthorizationFailure() throws Exception {
int maxBlockMs = 500;
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
index 97fe1913fa8..8d709666315 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
@@ -4196,137 +4196,6 @@ public class TransactionManagerTest {
assertFalse(transactionManager.hasOngoingTransaction());
}
- @Test
- public void testInitializeTransactionsWithKeepPreparedTxn() {
- doInitTransactionsWith2PCEnabled(true);
- runUntil(transactionManager::hasProducerId);
-
- // Expect a bumped epoch in the response.
- assertTrue(transactionManager.hasProducerId());
- assertFalse(transactionManager.hasOngoingTransaction());
- assertEquals(ongoingProducerId, transactionManager.producerIdAndEpoch().producerId);
- assertEquals(bumpedOngoingEpoch, transactionManager.producerIdAndEpoch().epoch);
- }
-
- @Test
- public void testPrepareTransaction() {
- doInitTransactionsWith2PCEnabled(false);
- runUntil(transactionManager::hasProducerId);
-
- // Begin a transaction
- transactionManager.beginTransaction();
- assertTrue(transactionManager.hasOngoingTransaction());
-
- // Add a partition to the transaction
- transactionManager.maybeAddPartition(tp0);
-
- // Capture the current producer ID and epoch before preparing the response
- long producerId = transactionManager.producerIdAndEpoch().producerId;
- short epoch = transactionManager.producerIdAndEpoch().epoch;
-
- // Simulate a produce request
- try {
- // Prepare the response before sending to ensure it's ready
- prepareProduceResponse(Errors.NONE, producerId, epoch);
-
- appendToAccumulator(tp0);
- // Wait until the request is processed
- runUntil(() -> !client.hasPendingResponses());
- } catch (InterruptedException e) {
- fail("Unexpected interruption: " + e);
- }
-
- transactionManager.prepareTransaction();
- assertTrue(transactionManager.isPrepared());
-
- ProducerIdAndEpoch preparedState = transactionManager.preparedTransactionState();
- // Validate the state contains the correct producer ID and epoch
- assertEquals(producerId, preparedState.producerId);
- assertEquals(epoch, preparedState.epoch);
- }
-
- @Test
- public void testInitPidResponseWithKeepPreparedTrueAndOngoingTransaction() {
- // Initialize transaction manager with 2PC enabled
- initializeTransactionManager(Optional.of(transactionalId), true, true);
-
- // Start initializeTransactions with keepPreparedTxn=true
- TransactionalRequestResult result = transactionManager.initializeTransactions(true);
-
- // Prepare coordinator response
- prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
- runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
-
- // Simulate InitProducerId response with ongoing transaction
- long ongoingPid = 12345L;
- short ongoingEpoch = 5;
- prepareInitPidResponse(
- Errors.NONE,
- false,
- producerId,
- epoch,
- true,
- true,
- ongoingPid,
- ongoingEpoch
- );
-
- runUntil(transactionManager::hasProducerId);
- transactionManager.maybeUpdateTransactionV2Enabled(true);
-
- result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, TEST_TIMEOUT_MSG);
- assertTrue(result.isSuccessful());
-
- // Verify transaction manager transitioned to PREPARED_TRANSACTION state
- assertTrue(transactionManager.isPrepared());
-
- // Verify preparedTxnState was set with ongoing producer ID and epoch
- ProducerIdAndEpoch preparedState = transactionManager.preparedTransactionState();
- assertNotNull(preparedState);
- assertEquals(ongoingPid, preparedState.producerId);
- assertEquals(ongoingEpoch, preparedState.epoch);
- }
-
- @Test
- public void testInitPidResponseWithKeepPreparedTrueAndNoOngoingTransaction() {
- // Initialize transaction manager without 2PC enabled
- // keepPrepared can be true even when enable2Pc is false, and we expect the same behavior
- initializeTransactionManager(Optional.of(transactionalId), true, false);
-
- // Start initializeTransactions with keepPreparedTxn=true
- TransactionalRequestResult result = transactionManager.initializeTransactions(true);
-
- // Prepare coordinator response
- prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
- runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
-
- // Simulate InitProducerId response without ongoing transaction
- prepareInitPidResponse(
- Errors.NONE,
- false,
- producerId,
- epoch,
- true,
- false,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_EPOCH
- );
-
- runUntil(transactionManager::hasProducerId);
- transactionManager.maybeUpdateTransactionV2Enabled(true);
-
- result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, TEST_TIMEOUT_MSG);
- assertTrue(result.isSuccessful());
-
- // Verify transaction manager transitioned to READY state (not PREPARED_TRANSACTION)
- assertFalse(transactionManager.isPrepared());
- assertTrue(transactionManager.isReady());
-
- // Verify preparedTxnState was not set or is empty
- ProducerIdAndEpoch preparedState = transactionManager.preparedTransactionState();
- assertEquals(ProducerIdAndEpoch.NONE, preparedState);
- }
-
private void prepareAddPartitionsToTxn(final Map<TopicPartition, Errors> errors) {
AddPartitionsToTxnResult result = AddPartitionsToTxnResponse.resultForTransaction(AddPartitionsToTxnResponse.V3_AND_BELOW_TXN_ID, errors);
AddPartitionsToTxnResponseData data = new AddPartitionsToTxnResponseData().setResultsByTopicV3AndBelow(result.topicResults()).setThrottleTimeMs(0);