Cumulative improvements from PRs #253-#260 (draft, diff review only)

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java

@@ -129,7 +129,7 @@ public static void pollUntilTrue(
         Consumer<byte[], byte[]> consumer,
         Duration timeout,
         Supplier<Boolean> testCondition,
-        long waitTimeMs, 
+        long waitTimeMs,
         String msg
     ) throws InterruptedException {
         TestUtils.waitForCondition(() -> {
@@ -472,7 +472,7 @@ public void onUpdate(ClusterResource clusterResource) {
         public byte[] serialize(String topic, byte[] data) {
             return data;
         }
-        
+
         public static void resetCount() {
             UPDATE_PRODUCER_COUNT.set(0);
         }

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/TransactionsExpirationTest.java

@@ -170,8 +170,7 @@ private void testTransactionAfterProducerIdExpires(ClusterInstance clusterInstan
 
         try (Producer<byte[], byte[]> producer = clusterInstance.producer(Map.of(
             ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTION_ID
-        ))
-        ) {
+        ))) {
             producer.initTransactions();
 
             // Start and then abort a transaction to allow the producer ID to expire.
@@ -208,8 +207,7 @@ private void testTransactionAfterProducerIdExpires(ClusterInstance clusterInstan
         // Create a new producer to check that we retain the producer ID in transactional state.
         try (Producer<byte[], byte[]> producer = clusterInstance.producer(Map.of(
             ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTION_ID
-        ))
-        ) {
+        ))) {
             producer.initTransactions();
 
             // Start a new transaction and attempt to send. This should work since only the producer ID was removed from its mapping in ProducerStateManager.

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/TransactionsWithMaxInFlightOneTest.java

@@ -84,8 +84,7 @@ public void testTransactionalProducerSingleBrokerMaxInFlightOne(ClusterInstance
         try (Producer<byte[], byte[]> producer = clusterInstance.producer(Map.of(
             ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional-producer",
             ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1
-        ))
-        ) {
+        ))) {
             producer.initTransactions();
 
             producer.beginTransaction();

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeProducersWithBrokerIdTest.java

@@ -44,7 +44,7 @@ public class DescribeProducersWithBrokerIdTest {
     private static final short REPLICATION_FACTOR = 3;
 
     private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC_NAME, 0);
-    
+
     private final ClusterInstance clusterInstance;
 
     public DescribeProducersWithBrokerIdTest(ClusterInstance clusterInstance) {
@@ -55,7 +55,7 @@ private static void sendTestRecords(Producer<byte[], byte[]> producer) {
         producer.send(new ProducerRecord<>(TOPIC_NAME, TOPIC_PARTITION.partition(), "key-0".getBytes(), "value-0".getBytes()));
         producer.flush();
     }
-    
+
     @BeforeEach
     void setUp() throws InterruptedException {
         clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
@@ -67,15 +67,15 @@ private List<Integer> getReplicaBrokerIds(Admin admin) throws Exception {
             .map(Node::id)
             .toList();
     }
-    
+
     private int getNonReplicaBrokerId(Admin admin) throws Exception {
         var replicaBrokerIds = getReplicaBrokerIds(admin);
         return clusterInstance.brokerIds().stream()
             .filter(id -> !replicaBrokerIds.contains(id))
             .findFirst()
             .orElseThrow(() -> new IllegalStateException("No non-replica broker found"));
     }
-    
+
     private int getFollowerBrokerId(Admin admin) throws Exception {
         var replicaBrokerIds = getReplicaBrokerIds(admin);
         var leaderBrokerId = clusterInstance.getLeaderBrokerId(TOPIC_PARTITION);
@@ -90,16 +90,16 @@ public void testDescribeProducersDefaultRoutesToLeader() throws Exception {
         try (Producer<byte[], byte[]> producer = clusterInstance.producer();
              var admin = clusterInstance.admin()) {
             sendTestRecords(producer);
-            
+
             var stateWithExplicitLeader = admin.describeProducers(
-                List.of(TOPIC_PARTITION), 
+                List.of(TOPIC_PARTITION),
                 new DescribeProducersOptions().brokerId(clusterInstance.getLeaderBrokerId(TOPIC_PARTITION))
             ).partitionResult(TOPIC_PARTITION).get();
-            
+
             var stateWithDefaultRouting = admin.describeProducers(
                 List.of(TOPIC_PARTITION)
             ).partitionResult(TOPIC_PARTITION).get();
-            
+
             assertNotNull(stateWithDefaultRouting);
             assertFalse(stateWithDefaultRouting.activeProducers().isEmpty());
             assertEquals(stateWithExplicitLeader.activeProducers(), stateWithDefaultRouting.activeProducers());
@@ -113,10 +113,10 @@ public void testDescribeProducersFromFollower() throws Exception {
             sendTestRecords(producer);
 
             var followerState = admin.describeProducers(
-                List.of(TOPIC_PARTITION), 
+                List.of(TOPIC_PARTITION),
                 new DescribeProducersOptions().brokerId(getFollowerBrokerId(admin))
             ).partitionResult(TOPIC_PARTITION).get();
-            
+
             var leaderState = admin.describeProducers(
                 List.of(TOPIC_PARTITION)
             ).partitionResult(TOPIC_PARTITION).get();
@@ -133,11 +133,11 @@ public void testDescribeProducersWithInvalidBrokerId() throws Exception {
              var admin = clusterInstance.admin()) {
             sendTestRecords(producer);
 
-            TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class, 
+            TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class,
                 admin.describeProducers(
-                    List.of(TOPIC_PARTITION), 
+                    List.of(TOPIC_PARTITION),
                     new DescribeProducersOptions().brokerId(getNonReplicaBrokerId(admin))
-                ).partitionResult(TOPIC_PARTITION)); 
+                ).partitionResult(TOPIC_PARTITION));
         }
     }
 }

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/StaticBrokerConfigTest.java

@@ -123,8 +123,8 @@ public void testTopicConfigsGetImpactedIfStaticConfigsAddToBroker(ClusterInstanc
     @ClusterTest(types = {Type.KRAFT})
     public void testInternalConfigsDoNotReturnForDescribeConfigs(ClusterInstance cluster) throws Exception {
         try (
-                Admin admin = cluster.admin();
-                Admin controllerAdmin = cluster.admin(Map.of(), true)
+            Admin admin = cluster.admin();
+            Admin controllerAdmin = cluster.admin(Map.of(), true)
         ) {
             ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
             ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC);

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerBounceTest.java

@@ -66,7 +66,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 
-import static  org.apache.kafka.test.TestUtils.SEEDED_RANDOM;
+import static org.apache.kafka.test.TestUtils.SEEDED_RANDOM;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerIntegrationTest.java

@@ -118,6 +118,7 @@ private static void testFetchPartitionsAfterFailedListener(ClusterInstance clust
                 ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name()))) {
             consumer.subscribe(List.of(topic), new ConsumerRebalanceListener() {
                 private int count = 0;
+
                 @Override
                 public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                 }

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCallbackTest.java

@@ -313,8 +313,8 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
             }
         });
         ClientsTestUtils.pollUntilTrue(
-            consumer, 
-            partitionsAssigned::get, 
+            consumer,
+            partitionsAssigned::get,
             "Timed out before expected rebalance completed"
         );
     }

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

@@ -340,6 +340,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                     // keep partitions paused in this test so that we can verify the commits based on specific seeks
                     consumer.pause(partitions);
                 }
+
                 @Override
                 public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                     // No-op

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerFetchTest.java

@@ -174,7 +174,7 @@ private void testFetchOutOfRangeOffsetResetConfigLatest(GroupProtocol groupProto
             sendRecords(producer, tp, totalRecords, startingTimestamp);
             consumer.assign(List.of(tp));
             consumer.seek(tp, 0);
-            
+
             // consume some, but not all the records
             consumeAndVerifyRecords(consumer, tp, totalRecords / 2, 0);
             // seek to out of range position
@@ -369,7 +369,7 @@ private void testFetchHonoursMaxPartitionFetchBytesIfLargeRecordNotFirst(GroupPr
     }
 
     private void checkFetchHonoursSizeIfLargeRecordNotFirst(
-        Map<String, Object> config, 
+        Map<String, Object> config,
         int largeProducerRecordSize
     ) throws ExecutionException, InterruptedException {
         try (Consumer<byte[], byte[]> consumer = cluster.consumer(config);
@@ -393,7 +393,7 @@ private void checkFetchHonoursSizeIfLargeRecordNotFirst(
 
             // we should only get the small record in the first `poll`
             consumer.assign(List.of(tp));
-            
+
             var records = consumeRecords(consumer, 1);
             assertEquals(1, records.size());
             var consumerRecord = records.iterator().next();
@@ -426,7 +426,7 @@ private void testLowMaxFetchSizeForRequestAndPartition(GroupProtocol groupProtoc
             // Avoid a rebalance while the records are being sent (the default is 6 seconds)
             MAX_POLL_INTERVAL_MS_CONFIG, 20000
         );
-        
+
         try (Consumer<byte[], byte[]> consumer = cluster.consumer(config);
              Producer<byte[], byte[]> producer = cluster.producer()
         ) {

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerSubscriptionTest.java

@@ -327,7 +327,7 @@ public void testAsyncConsumerRe2JPatternSubscription() throws InterruptedExcepti
                 new TopicPartition(topic1, 1)
             );
             awaitAssignment(consumer, assignment);
-            
+
             consumer.unsubscribe();
             assertEquals(0, consumer.assignment().size());
             // Subscribe to a different pattern to match topic2 (that did not match before)
@@ -385,7 +385,7 @@ public void testAsyncConsumerRe2JPatternExpandSubscription() throws InterruptedE
 
             var topic2 = "topic2"; // does not match first pattern
             cluster.createTopic(topic2, 2, (short) BROKER_COUNT);
-            
+
             assertEquals(0, consumer.assignment().size());
             var pattern = new SubscriptionPattern("topic1.*");
             consumer.subscribe(pattern);
@@ -418,8 +418,8 @@ public void testTopicIdSubscriptionWithRe2JRegexAndOffsetsFetch() throws Interru
 
         Map<String, Object> config = Map.of(GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT));
         try (
-                Producer<byte[], byte[]> producer = cluster.producer();
-                Consumer<byte[], byte[]> consumer = cluster.consumer(config)
+            Producer<byte[], byte[]> producer = cluster.producer();
+            Consumer<byte[], byte[]> consumer = cluster.consumer(config)
         ) {
             assertEquals(0, consumer.assignment().size());
 
@@ -504,7 +504,7 @@ public void testRe2JPatternSubscriptionAndTopicSubscription() throws Interrupted
             awaitAssignment(consumer, patternAssignment);
         }
     }
-    
+
 
     @ClusterTest
     public void testRe2JPatternSubscriptionInvalidRegex() throws InterruptedException {