KAFKA-19042 Rewrite TransactionsBounceTest from Scala to Java (#23879)

Rewrite TransactionsBounceTest from Scala to Java and move it to clients-integration-tests. The transaction status header helpers are moved to ClientsTestUtils and shared with ProducerIdExpirationTest, and the Scala helpers in TestUtils that were only used by this test are removed. Reviewers: Murali Basani <muralidhar.basani@aiven.io>, Chia-Ping Tsai <chia7712@gmail.com>
2026-10-28 08:24:58 +08:00 · 2026-10-28 05:05:05 +08:00
1 parent b1093f64c0
commit 7596c4bc4b
@@ -29,6 +29,8 @@ import org.apache.kafka.common.ClusterResource;
 import org.apache.kafka.common.ClusterResourceListener;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.serialization.Deserializer;
@@ -36,6 +38,7 @@ import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.test.TestUtils;

+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -53,6 +56,7 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_
 import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -62,6 +66,9 @@ public class ClientsTestUtils {

    private static final String KEY_PREFIX = "key ";
    private static final String VALUE_PREFIX = "value ";
+    private static final String TRANSACTION_STATUS_HEADER_KEY = "transactionStatus";
+    private static final byte[] COMMITTED_VALUE = "committed".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] ABORTED_VALUE = "aborted".getBytes(StandardCharsets.UTF_8);

    private ClientsTestUtils() {}

@@ -369,6 +376,16 @@ public class ClientsTestUtils {
        }, "Continuous poll not fail");
    }

+    public static Header transactionStatusHeader(boolean willBeCommitted) {
+        return new RecordHeader(TRANSACTION_STATUS_HEADER_KEY, willBeCommitted ? COMMITTED_VALUE : ABORTED_VALUE);
+    }
+
+    public static void assertTransactionCommitted(ConsumerRecord<?, ?> record) {
+        Header header = record.headers().lastHeader(TRANSACTION_STATUS_HEADER_KEY);
+        assertNotNull(header, "Expected transaction status header");
+        assertArrayEquals(COMMITTED_VALUE, header.value(), "Expected committed status");
+    }
+
    /**
     * This class is intended to replace the test cases in BaseConsumerTest.scala.
     * When converting tests that extend from BaseConsumerTest.scala to Java,
@@ -30,7 +30,6 @@ import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.InvalidPidMappingException;
 import org.apache.kafka.common.errors.TransactionalIdNotFoundException;
-import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
@@ -49,7 +48,9 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;

+import static org.apache.kafka.clients.ClientsTestUtils.assertTransactionCommitted;
 import static org.apache.kafka.clients.ClientsTestUtils.consumeRecords;
+import static org.apache.kafka.clients.ClientsTestUtils.transactionStatusHeader;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.ISOLATION_LEVEL_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG;
 import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
@@ -68,11 +69,9 @@ import static org.apache.kafka.server.config.ServerLogConfigs.AUTO_CREATE_TOPICS
 import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
 import static org.apache.kafka.test.TestUtils.assertFutureThrows;
 import static org.apache.kafka.test.TestUtils.waitForCondition;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;

@ClusterTestDefaults(
    brokers = 3,
@@ -100,10 +99,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
    }
 )
 public class ProducerIdExpirationTest {
-    private static final String HEADER_KEY = "transactionStatus";
-    private static final byte[] COMMITTED_VALUE = "committed".getBytes(StandardCharsets.UTF_8);
-    private static final byte[] ABORTED_VALUE = "aborted".getBytes(StandardCharsets.UTF_8);
-
    private final String topic1 = "topic1";
    private final int numPartitions = 1;
    private final short replicationFactor = 3;
@@ -310,18 +305,14 @@ public class ProducerIdExpirationTest {
        String value,
        boolean willBeCommitted
    ) {
-        var header = new RecordHeader(HEADER_KEY, willBeCommitted ? COMMITTED_VALUE : ABORTED_VALUE);
        return new ProducerRecord<>(topic, partition,
            key.getBytes(StandardCharsets.UTF_8),
            value.getBytes(StandardCharsets.UTF_8),
-            Collections.singleton(header));
+            Collections.singleton(transactionStatusHeader(willBeCommitted)));
    }

    private static String assertCommittedAndGetValue(ConsumerRecord<byte[], byte[]> record) {
-        var headers = record.headers().headers(HEADER_KEY).iterator();
-        assertTrue(headers.hasNext(), "Expected transaction status header");
-        var header = headers.next();
-        assertArrayEquals(COMMITTED_VALUE, header.value(), "Expected committed status");
+        assertTransactionCommitted(record);
        return new String(record.value(), StandardCharsets.UTF_8);
    }

@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.producer;
+
+import org.apache.kafka.clients.ClientsTestUtils;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.test.TestUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@ClusterTestDefaults(
+    types = {Type.KRAFT},
+    brokers = 4,
+    serverProperties = {
+        @ClusterConfigProperty(key = ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "false"),
+        @ClusterConfigProperty(key = ServerConfigs.MESSAGE_MAX_BYTES_CONFIG, value = "32768"),
+        @ClusterConfigProperty(key = ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "true"),
+        @ClusterConfigProperty(key = ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG, value = "false"),
+        // Use fewer partitions for __consumer_offsets so topic creation and leader assignment do not take too long.
+        @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+        @ClusterConfigProperty(key = TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, value = "2"),
+        // Allow a small session timeout.
+        @ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "10"),
+        @ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0"),
+        @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+    }
+)
+public class TransactionsBounceTest {
+    private static final int CONSUME_RECORD_TIMEOUT_MS = 30000;
+    private static final int NUM_PARTITIONS = 3;
+    private static final String OUTPUT_TOPIC = "output-topic";
+    private static final String INPUT_TOPIC = "input-topic";
+
+    private final ClusterInstance cluster;
+
+    TransactionsBounceTest(ClusterInstance cluster) {
+        this.cluster = cluster;
+    }
+
+    @ClusterTest
+    public void testWithGroupMetadataClassic() throws Exception {
+        testBrokerFailure(GroupProtocol.CLASSIC);
+    }
+
+    @ClusterTest
+    public void testWithGroupMetadataConsumer() throws Exception {
+        testBrokerFailure(GroupProtocol.CONSUMER);
+    }
+
+    private void testBrokerFailure(GroupProtocol groupProtocol) throws Exception {
+        // Seed a topic with 10000 records and copy it transactionally while constantly bouncing brokers.
+        int numInputRecords = 10000;
+        Map<String, String> topicConfigs = Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
+        cluster.createTopic(INPUT_TOPIC, NUM_PARTITIONS, (short) 3, topicConfigs);
+        cluster.createTopic(OUTPUT_TOPIC, NUM_PARTITIONS, (short) 3, topicConfigs);
+
+        try (Producer<byte[], byte[]> producer = cluster.producer()) {
+            for (int i = 0; i < numInputRecords; i++) {
+                byte[] value = Integer.toString(i).getBytes(StandardCharsets.UTF_8);
+                producer.send(new ProducerRecord<>(INPUT_TOPIC, value, value));
+            }
+            producer.flush();
+        }
+
+        try (Consumer<byte[], byte[]> consumer = createConsumerAndSubscribe("myGroup", INPUT_TOPIC, groupProtocol, false);
+             Producer<byte[], byte[]> producer = cluster.producer(Map.of(
+                 ProducerConfig.BATCH_SIZE_CONFIG, 512,
+                 ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-txn"
+             ))) {
+            producer.initTransactions();
+            AtomicBoolean bouncing = new AtomicBoolean(true);
+            CompletableFuture<Void> bouncer = CompletableFuture.runAsync(() -> bounceBrokers(bouncing));
+            try {
+                int numMessagesProcessed = 0;
+                int iteration = 0;
+                while (numMessagesProcessed < numInputRecords) {
+                    int toRead = Math.min(200, numInputRecords - numMessagesProcessed);
+                    List<ConsumerRecord<byte[], byte[]>> records = ClientsTestUtils.consumeRecords(
+                        consumer, toRead, Integer.MAX_VALUE, CONSUME_RECORD_TIMEOUT_MS);
+
+                    producer.beginTransaction();
+                    boolean shouldAbort = iteration % 3 == 0;
+                    for (ConsumerRecord<byte[], byte[]> record : records) {
+                        ProducerRecord<byte[], byte[]> outputRecord = new ProducerRecord<>(OUTPUT_TOPIC, record.key(), record.value());
+                        // Mark the intended outcome so the verifying consumer can detect any aborted records it receives.
+                        outputRecord.headers().add(ClientsTestUtils.transactionStatusHeader(!shouldAbort));
+                        producer.send(outputRecord, new ErrorLoggingCallback(OUTPUT_TOPIC, record.key(), record.value(), true));
+                    }
+
+                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+                    for (TopicPartition partition : consumer.assignment()) {
+                        offsets.put(partition, new OffsetAndMetadata(consumer.position(partition)));
+                    }
+                    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
+
+                    if (shouldAbort) {
+                        producer.abortTransaction();
+                        resetToCommittedPositions(consumer);
+                    } else {
+                        producer.commitTransaction();
+                        numMessagesProcessed += records.size();
+                    }
+                    iteration++;
+                }
+            } finally {
+                bouncing.set(false);
+                bouncer.get();
+            }
+        }
+
+        try (Consumer<byte[], byte[]> consumer = createConsumerAndSubscribe("randomGroup", OUTPUT_TOPIC, groupProtocol, true)) {
+            Map<TopicPartition, List<Integer>> recordsByPartition = new HashMap<>();
+            List<ConsumerRecord<byte[], byte[]>> records = ClientsTestUtils.consumeRecords(
+                consumer, numInputRecords, Integer.MAX_VALUE, CONSUME_RECORD_TIMEOUT_MS);
+            for (ConsumerRecord<byte[], byte[]> record : records) {
+                ClientsTestUtils.assertTransactionCommitted(record);
+                TopicPartition partition = new TopicPartition(record.topic(), record.partition());
+                recordsByPartition.computeIfAbsent(partition, __ -> new ArrayList<>())
+                    .add(Integer.parseInt(new String(record.value(), StandardCharsets.UTF_8)));
+            }
+
+            Set<Integer> recordSet = new HashSet<>();
+            for (List<Integer> partitionValues : recordsByPartition.values()) {
+                assertEquals(partitionValues.stream().sorted().collect(Collectors.toList()), partitionValues,
+                    "Out of order messages detected");
+                recordSet.addAll(partitionValues);
+            }
+            assertEquals(numInputRecords, records.size(), "Unexpected number of committed records");
+            assertEquals(numInputRecords, recordSet.size());
+            assertEquals(IntStream.range(0, numInputRecords).boxed().collect(Collectors.toSet()), recordSet);
+        }
+    }
+
+    private Consumer<byte[], byte[]> createConsumerAndSubscribe(
+        String groupId,
+        String topic,
+        GroupProtocol groupProtocol,
+        boolean readCommitted
+    ) {
+        Consumer<byte[], byte[]> consumer = cluster.consumer(Map.of(
+            ConsumerConfig.GROUP_ID_CONFIG, groupId,
+            ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name(),
+            ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,
+            ConsumerConfig.ISOLATION_LEVEL_CONFIG, readCommitted ? "read_committed" : "read_uncommitted"
+        ));
+        consumer.subscribe(List.of(topic));
+        return consumer;
+    }
+
+    private void resetToCommittedPositions(Consumer<byte[], byte[]> consumer) {
+        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(consumer.assignment());
+        for (TopicPartition partition : consumer.assignment()) {
+            OffsetAndMetadata offset = committed.get(partition);
+            if (offset == null) {
+                consumer.seekToBeginning(List.of(partition));
+            } else {
+                consumer.seek(partition, offset.offset());
+            }
+        }
+    }
+
+    private void bounceBrokers(AtomicBoolean bouncing) {
+        try (Admin admin = cluster.admin()) {
+            while (bouncing.get()) {
+                for (int brokerId : cluster.brokerIds()) {
+                    cluster.shutdownBroker(brokerId);
+                    cluster.brokers().get(brokerId).awaitShutdown();
+                    TimeUnit.MILLISECONDS.sleep(500);
+                    cluster.startBroker(brokerId);
+                    TimeUnit.MILLISECONDS.sleep(500);
+                }
+                for (int partition = 0; partition < NUM_PARTITIONS; partition++) {
+                    TopicPartition topicPartition = new TopicPartition(OUTPUT_TOPIC, partition);
+                    TestUtils.waitForCondition(() -> hasLeader(admin, topicPartition), "No leader elected for " + topicPartition);
+                }
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Broker bouncer failed", e);
+        }
+    }
+
+    private boolean hasLeader(Admin admin, TopicPartition topicPartition) throws Exception {
+        TopicPartitionInfo info = admin.describeTopics(List.of(topicPartition.topic()))
+            .allTopicNames().get().get(topicPartition.topic())
+            .partitions().get(topicPartition.partition());
+        return info.leader() != null && info.leader().id() >= 0;
+    }
+}
@@ -1,207 +0,0 @@
-/**
-  * Licensed to the Apache Software Foundation (ASF) under one or more
-  * contributor license agreements.  See the NOTICE file distributed with
-  * this work for additional information regarding copyright ownership.
-  * The ASF licenses this file to You under the Apache License, Version 2.0
-  * (the "License"); you may not use this file except in compliance with
-  * the License.  You may obtain a copy of the License at
-  *
-  *    http://www.apache.org/licenses/LICENSE-2.0
-  *
-  * Unless required by applicable law or agreed to in writing, software
-  * distributed under the License is distributed on an "AS IS" BASIS,
-  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  * See the License for the specific language governing permissions and
-  * limitations under the License.
-  */
-
-package kafka.api
-
-import java.util
-import java.util.Properties
-import kafka.server.KafkaConfig
-import kafka.utils.{TestInfoUtils, TestUtils}
-import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig}
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig}
-import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
-import org.apache.kafka.coordinator.transaction.TransactionLogConfig
-import org.apache.kafka.server.util.ShutdownableThread
-import org.apache.kafka.server.config.{ReplicationConfigs, ServerConfigs, ServerLogConfigs}
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.MethodSource
-
-import scala.jdk.CollectionConverters._
-import scala.collection.mutable
-
-class TransactionsBounceTest extends IntegrationTestHarness {
-  private val consumeRecordTimeout = 30000
-  private val producerBufferSize =  65536
-  private val serverMessageMaxBytes =  producerBufferSize/2
-  private val numPartitions = 3
-  private val outputTopic = "output-topic"
-  private val inputTopic = "input-topic"
-
-  val overridingProps = new Properties()
-  overridingProps.put(ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, false.toString)
-  overridingProps.put(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG, serverMessageMaxBytes.toString)
-  // Set a smaller value for the number of partitions for the offset commit topic (__consumer_offset topic)
-  // so that the creation of that topic/partition(s) and subsequent leader assignment doesn't take relatively long
-  overridingProps.put(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, true.toString)
-  overridingProps.put(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, false.toString)
-  overridingProps.put(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG, false.toString)
-  overridingProps.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, 1.toString)
-  overridingProps.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, 3.toString)
-  overridingProps.put(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, 2.toString)
-  overridingProps.put(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, "10") // set small enough session timeout
-  overridingProps.put(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, "0")
-  overridingProps.put(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, 1.toString)
-  overridingProps.put(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, 3.toString)
-
-  // This is the one of the few tests we currently allow to preallocate ports, despite the fact that this can result in transient
-  // failures due to ports getting reused. We can't use random ports because of bad behavior that can result from bouncing
-  // brokers too quickly when they get new, random ports. If we're not careful, the client can end up in a situation
-  // where metadata is not refreshed quickly enough, and by the time it's actually trying to, all the servers have
-  // been bounced and have new addresses. None of the bootstrap nodes or current metadata can get them connected to a
-  // running server.
-  //
-  // Since such quick rotation of servers is incredibly unrealistic, we allow this one test to preallocate ports, leaving
-  // a small risk of hitting errors due to port conflicts. Hopefully this is infrequent enough to not cause problems.
-  override def generateConfigs = {
-    FixedPortTestUtils.createBrokerConfigs(brokerCount)
-      .map(KafkaConfig.fromProps(_, overridingProps))
-  }
-
-  override protected def brokerCount: Int = 4
-
-  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
-  @MethodSource(Array("getTestGroupProtocolParametersAll"))
-  def testWithGroupMetadata(groupProtocol: String): Unit = {
-    testBrokerFailure((producer, _, consumer) =>
-      producer.sendOffsetsToTransaction(TestUtils.consumerPositions(consumer).asJava, consumer.groupMetadata()))
-  }
-
-  private def testBrokerFailure(commit: (KafkaProducer[Array[Byte], Array[Byte]],
-    String, Consumer[Array[Byte], Array[Byte]]) => Unit): Unit = {
-    // basic idea is to seed a topic with 10000 records, and copy it transactionally while bouncing brokers
-    // constantly through the period.
-    val consumerGroup = "myGroup"
-    val numInputRecords = 10000
-    createTopics()
-
-    TestUtils.seedTopicWithNumberedRecords(inputTopic, numInputRecords, brokers)
-    val consumer = createConsumerAndSubscribe(consumerGroup, List(inputTopic))
-    val producer = createTransactionalProducer("test-txn")
-
-    producer.initTransactions()
-
-    val scheduler = new BounceScheduler
-    scheduler.start()
-
-    try {
-      var numMessagesProcessed = 0
-      var iteration = 0
-
-      while (numMessagesProcessed < numInputRecords) {
-        val toRead = Math.min(200, numInputRecords - numMessagesProcessed)
-        trace(s"$iteration: About to read $toRead messages, processed $numMessagesProcessed so far..")
-        val records = TestUtils.pollUntilAtLeastNumRecords(consumer, toRead, waitTimeMs = consumeRecordTimeout)
-        trace(s"Received ${records.size} messages, sending them transactionally to $outputTopic")
-
-        producer.beginTransaction()
-        val shouldAbort = iteration % 3 == 0
-        records.foreach { record =>
-          producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(outputTopic, null, record.key, record.value, !shouldAbort), new ErrorLoggingCallback(outputTopic, record.key, record.value, true))
-        }
-        trace(s"Sent ${records.size} messages. Committing offsets.")
-        commit(producer, consumerGroup, consumer)
-
-        if (shouldAbort) {
-          trace(s"Committed offsets. Aborting transaction of ${records.size} messages.")
-          producer.abortTransaction()
-          TestUtils.resetToCommittedPositions(consumer)
-        } else {
-          trace(s"Committed offsets. committing transaction of ${records.size} messages.")
-          producer.commitTransaction()
-          numMessagesProcessed += records.size
-        }
-        iteration += 1
-      }
-    } finally {
-      scheduler.shutdown()
-    }
-
-    val verifyingConsumer = createConsumerAndSubscribe("randomGroup", List(outputTopic), readCommitted = true)
-    val recordsByPartition = new mutable.HashMap[TopicPartition, mutable.ListBuffer[Int]]()
-    TestUtils.pollUntilAtLeastNumRecords(verifyingConsumer, numInputRecords, waitTimeMs = consumeRecordTimeout).foreach { record =>
-      val value = TestUtils.assertCommittedAndGetValue(record).toInt
-      val topicPartition = new TopicPartition(record.topic(), record.partition())
-      recordsByPartition.getOrElseUpdate(topicPartition, new mutable.ListBuffer[Int])
-        .append(value)
-    }
-
-    val outputRecords = new mutable.ListBuffer[Int]()
-    recordsByPartition.values.foreach { partitionValues =>
-      assertEquals(partitionValues, partitionValues.sorted, "Out of order messages detected")
-      outputRecords.appendAll(partitionValues)
-    }
-
-    val recordSet = outputRecords.toSet
-    assertEquals(numInputRecords, recordSet.size)
-
-    val expectedValues = (0 until numInputRecords).toSet
-    assertEquals(expectedValues, recordSet, s"Missing messages: ${expectedValues -- recordSet}")
-  }
-
-  private def createTransactionalProducer(transactionalId: String) = {
-    val props = new Properties()
-    props.put(ProducerConfig.ACKS_CONFIG, "all")
-    props.put(ProducerConfig.BATCH_SIZE_CONFIG, "512")
-    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId)
-    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
-    createProducer(configOverrides = props)
-  }
-
-  private def createConsumerAndSubscribe(groupId: String,
-                                         topics: List[String],
-                                         readCommitted: Boolean = false) = {
-    val consumerProps = new Properties
-    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
-    consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
-    consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
-      if (readCommitted) "read_committed" else "read_uncommitted")
-    val consumer = createConsumer(configOverrides = consumerProps)
-    consumer.subscribe(topics.asJava)
-    consumer
-  }
-
-  private def createTopics() =  {
-    val topicConfigs = util.Map.of(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, 2.toString)
-    createTopic(inputTopic, numPartitions, 3, topicConfigs)
-    createTopic(outputTopic, numPartitions, 3, topicConfigs)
-  }
-
-  private class BounceScheduler extends ShutdownableThread("daemon-broker-bouncer", false) {
-    override def doWork(): Unit = {
-      for (server <- brokers) {
-        trace("Shutting down server : %s".format(server.config.brokerId))
-        server.shutdown()
-        server.awaitShutdown()
-        Thread.sleep(500)
-        trace("Server %s shut down. Starting it up again.".format(server.config.brokerId))
-        server.startup()
-        trace("Restarted server: %s".format(server.config.brokerId))
-        Thread.sleep(500)
-      }
-
-      (0 until numPartitions).foreach(partition => TestUtils.waitUntilLeaderIsElectedOrChangedWithAdmin(createAdminClient(), outputTopic, partition))
-    }
-
-    override def shutdown(): Unit = {
-      super.shutdown()
-   }
-  }
-
-}
@@ -28,7 +28,6 @@ import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter
 import org.apache.kafka.common.compress.Compression
 import org.apache.kafka.common.config.{ConfigException, ConfigResource}
 import org.apache.kafka.common.errors.{TopicExistsException, UnknownTopicOrPartitionException}
-import org.apache.kafka.common.header.Header
 import org.apache.kafka.common.internals.{Plugin, Topic}
 import org.apache.kafka.common.network.{ConnectionMode, ListenerName}
 import org.apache.kafka.common.record.TimestampType
@@ -58,14 +57,13 @@ import org.mockito.Mockito

 import java.io._
 import java.nio._
-import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, StandardOpenOption}
 import java.time.Duration
 import java.util
 import java.util.concurrent._
 import java.util.{Optional, Properties}
 import scala.collection.mutable.ArrayBuffer
-import scala.collection.{Map, Seq, mutable}
+import scala.collection.{Map, Seq}
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.jdk.CollectionConverters._
@@ -89,9 +87,6 @@ object TestUtils extends Logging {
  // CN in SSL certificates - this is used for endpoint validation when enabled
  val SslCertificateCn = "localhost"

-  private val transactionStatusKey = "transactionStatus"
-  private val committedValue : Array[Byte] = "committed".getBytes(StandardCharsets.UTF_8)
-
  sealed trait LogDirFailureType
  case object Roll extends LogDirFailureType
  case object Checkpoint extends LogDirFailureType
@@ -1066,77 +1061,6 @@ object TestUtils extends Logging {
    new KafkaProducer[Array[Byte], Array[Byte]](props, new ByteArraySerializer, new ByteArraySerializer)
  }

-  // Seeds the given topic with records with keys and values in the range [0..numRecords)
-  def seedTopicWithNumberedRecords[B <: KafkaBroker](
-      topic: String,
-      numRecords: Int,
-      brokers: Seq[B]): Unit = {
-    val props = new Properties()
-    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
-    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, plaintextBootstrapServers(brokers))
-    val producer = new KafkaProducer[Array[Byte], Array[Byte]](props, new ByteArraySerializer, new ByteArraySerializer)
-    try {
-      for (i <- 0 until numRecords) {
-        producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, asBytes(i.toString), asBytes(i.toString)))
-      }
-      producer.flush()
-    } finally {
-      producer.close()
-    }
-  }
-
-  private def asString(bytes: Array[Byte]) = new String(bytes, StandardCharsets.UTF_8)
-
-  private def asBytes(string: String) = string.getBytes(StandardCharsets.UTF_8)
-
-  // Verifies that the record was intended to be committed by checking the headers for an expected transaction status
-  // If true, this will return the value as a string. It is expected that the record in question should have been created
-  // by the `producerRecordWithExpectedTransactionStatus` method.
-  def assertCommittedAndGetValue(record: ConsumerRecord[Array[Byte], Array[Byte]]) : String = {
-    record.headers.headers(transactionStatusKey).asScala.headOption match {
-      case Some(header) =>
-        assertEquals(asString(committedValue), asString(header.value), s"Got ${asString(header.value)} but expected the value to indicate " +
-          s"committed status.")
-      case None =>
-        fail("expected the record header to include an expected transaction status, but received nothing.")
-    }
-    recordValueAsString(record)
-  }
-
-  def recordValueAsString(record: ConsumerRecord[Array[Byte], Array[Byte]]) : String = {
-    asString(record.value)
-  }
-
-  def producerRecordWithExpectedTransactionStatus(topic: String, partition: Integer, key: Array[Byte], value: Array[Byte], willBeCommitted: Boolean): ProducerRecord[Array[Byte], Array[Byte]] = {
-    val header = new Header {override def key() = transactionStatusKey
-      override def value() = if (willBeCommitted)
-        committedValue
-      else
-        "aborted".getBytes(StandardCharsets.UTF_8)
-    }
-    new ProducerRecord[Array[Byte], Array[Byte]](topic, partition, key, value, util.Set.of(header))
-  }
-
-  // Collect the current positions for all partition in the consumers current assignment.
-  def consumerPositions(consumer: Consumer[Array[Byte], Array[Byte]]) : Map[TopicPartition, OffsetAndMetadata]  = {
-    val offsetsToCommit = new mutable.HashMap[TopicPartition, OffsetAndMetadata]()
-    consumer.assignment.forEach { topicPartition =>
-      offsetsToCommit.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition)))
-    }
-    offsetsToCommit.toMap
-  }
-
-  def resetToCommittedPositions(consumer: Consumer[Array[Byte], Array[Byte]]): Unit = {
-    val committed = consumer.committed(consumer.assignment).asScala.filter(_._2 != null).map { case (k, v) => k -> v.offset }
-
-    consumer.assignment.forEach { topicPartition =>
-      if (committed.contains(topicPartition))
-        consumer.seek(topicPartition, committed(topicPartition))
-      else
-        consumer.seekToBeginning(util.List.of(topicPartition))
-    }
-  }
-
  def incrementalAlterConfigs[B <: KafkaBroker](
      servers: Seq[B],
      adminClient: Admin,