MINOR: Various cleanups in clients tests

checkstyle/suppressions.xml

@@ -56,7 +56,7 @@
     <suppress checks="ClassFanOutComplexity"
               files="(SaslServerAuthenticator|SaslAuthenticatorTest).java"/>
     <suppress checks="NPath"
-              files="SaslServerAuthenticator.java"/>
+              files="(Microbenchmarks|SaslServerAuthenticator).java"/>
     <suppress checks="ClassFanOutComplexity"
               files="Errors.java"/>
     <suppress checks="ClassFanOutComplexity"

clients/src/test/java/org/apache/kafka/clients/AddressChangeHostResolver.java

@@ -19,9 +19,9 @@
 import java.net.InetAddress;
 
 class AddressChangeHostResolver implements HostResolver {
+    private final InetAddress[] initialAddresses;
+    private final InetAddress[] newAddresses;
     private boolean useNewAddresses;
-    private InetAddress[] initialAddresses;
-    private InetAddress[] newAddresses;
     private int resolutionCount = 0;
 
     public AddressChangeHostResolver(InetAddress[] initialAddresses, InetAddress[] newAddresses) {

clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java

@@ -58,7 +58,7 @@ public void testParseAndValidateAddressesWithReverseLookup() {
         // With lookup of example.com, either one or two addresses are expected depending on
         // whether ipv4 and ipv6 are enabled
         List<InetSocketAddress> validatedAddresses = checkWithLookup(asList("example.com:10000"));
-        assertTrue(validatedAddresses.size() >= 1, "Unexpected addresses " + validatedAddresses);
+        assertFalse(validatedAddresses.isEmpty(), "Unexpected addresses " + validatedAddresses);
         List<String> validatedHostNames = validatedAddresses.stream().map(InetSocketAddress::getHostName)
                 .collect(Collectors.toList());
         List<String> expectedHostNames = asList("93.184.215.14", "2606:2800:21f:cb07:6820:80da:af6b:8b2c");

clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java

@@ -75,15 +75,16 @@ public class ClusterConnectionStatesTest {
     private final String nodeId2 = "2002";
     private final String nodeId3 = "3003";
     private final String hostTwoIps = "multiple.ip.address";
-    private ClusterConnectionStates connectionStates;
 
     // For testing nodes with a single IP address, use localhost and default DNS resolution
-    private DefaultHostResolver singleIPHostResolver = new DefaultHostResolver();
+    private final DefaultHostResolver singleIPHostResolver = new DefaultHostResolver();
 
     // For testing nodes with multiple IP addresses, mock DNS resolution to get consistent results
-    private AddressChangeHostResolver multipleIPHostResolver = new AddressChangeHostResolver(
+    private final AddressChangeHostResolver multipleIPHostResolver = new AddressChangeHostResolver(
             initialAddresses.toArray(new InetAddress[0]), newAddresses.toArray(new InetAddress[0]));
 
+    private ClusterConnectionStates connectionStates;
+
     @BeforeEach
     public void setup() {
         this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax,

clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java

@@ -52,6 +52,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -67,12 +68,7 @@ public class FetchSessionHandlerTest {
      * ordering for test purposes.
      */
     private static Set<TopicPartition> toSet(TopicPartition... arr) {
-        TreeSet<TopicPartition> set = new TreeSet<>(new Comparator<TopicPartition>() {
-            @Override
-            public int compare(TopicPartition o1, TopicPartition o2) {
-                return o1.toString().compareTo(o2.toString());
-            }
-        });
+        TreeSet<TopicPartition> set = new TreeSet<>(Comparator.comparing(TopicPartition::toString));
         set.addAll(Arrays.asList(arr));
         return set;
     }
@@ -317,12 +313,7 @@ public void testDoubleBuild() {
         builder.add(new TopicPartition("foo", 0),
             new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 100, 200, Optional.empty()));
         builder.build();
-        try {
-            builder.build();
-            fail("Expected calling build twice to fail.");
-        } catch (Throwable t) {
-            // expected
-        }
+        assertThrows(Throwable.class, builder::build, "Expected calling build twice to fail.");
     }
 
     @Test

clients/src/test/java/org/apache/kafka/clients/InFlightRequestsTest.java

@@ -34,9 +34,9 @@
 
 public class InFlightRequestsTest {
 
+    private final String dest = "dest";
     private InFlightRequests inFlightRequests;
     private int correlationId;
-    private String dest = "dest";
 
     @BeforeEach
     public void setup() {

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

@@ -73,16 +73,17 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 public class MetadataTest {
 
-    private long refreshBackoffMs = 100;
-    private long refreshBackoffMaxMs = 1000;
-    private long metadataExpireMs = 1000;
+    private final long refreshBackoffMs = 100;
+    private final long refreshBackoffMaxMs = 1000;
+    private final long metadataExpireMs = 1000;
     private Metadata metadata = new Metadata(refreshBackoffMs, refreshBackoffMaxMs,
             metadataExpireMs, new LogContext(), new ClusterResourceListeners());
 
@@ -1210,8 +1211,7 @@ else if (partition.equals(internalPart))
         metadata.update(versionAndBuilder.requestVersion,
             RequestTestUtils.metadataUpdateWith(clusterId, numNodes, errorCounts, topicPartitionCounts, tp -> null, metadataSupplier, ApiKeys.METADATA.latestVersion(), topicIds),
             false, time.milliseconds());
-        List<Node> nodes = new ArrayList<>();
-        nodes.addAll(metadata.fetch().nodes());
+        List<Node> nodes = new ArrayList<>(metadata.fetch().nodes());
         Node controller = metadata.fetch().controller();
         assertEquals(numNodes, nodes.size());
         assertFalse(metadata.updateRequested());
@@ -1323,7 +1323,7 @@ public void testConcurrentUpdateAndFetchForSnapshotAndCluster() throws Interrupt
                 } else { // Thread to read metadata snapshot, once its updated
                     try {
                         if (!atleastMetadataUpdatedOnceLatch.await(5, TimeUnit.MINUTES)) {
-                            assertFalse(true, "Test had to wait more than 5 minutes, something went wrong.");
+                            fail("Test had to wait more than 5 minutes, something went wrong.");
                         }
                     } catch (InterruptedException e) {
                         throw new RuntimeException(e);
@@ -1335,7 +1335,7 @@ public void testConcurrentUpdateAndFetchForSnapshotAndCluster() throws Interrupt
             });
         }
         if (!allThreadsDoneLatch.await(5, TimeUnit.MINUTES)) {
-            assertFalse(true, "Test had to wait more than 5 minutes, something went wrong.");
+            fail("Test had to wait more than 5 minutes, something went wrong.");
         }
 
         // Validate new snapshot is upto-date. And has higher partition counts, nodes & leader epoch than earlier.

clients/src/test/java/org/apache/kafka/clients/admin/TopicCollectionTest.java

@@ -34,11 +34,11 @@ public void testTopicCollection() {
         List<Uuid> topicIds = Arrays.asList(Uuid.randomUuid(), Uuid.randomUuid(), Uuid.randomUuid());
         List<String> topicNames = Arrays.asList("foo", "bar");
 
-        TopicCollection idCollection = TopicCollection.ofTopicIds(topicIds);
-        TopicCollection nameCollection = TopicCollection.ofTopicNames(topicNames);
+        TopicIdCollection idCollection = TopicCollection.ofTopicIds(topicIds);
+        TopicNameCollection nameCollection = TopicCollection.ofTopicNames(topicNames);
 
-        assertTrue(((TopicIdCollection) idCollection).topicIds().containsAll(topicIds));
-        assertTrue(((TopicNameCollection) nameCollection).topicNames().containsAll(topicNames));
+        assertTrue(idCollection.topicIds().containsAll(topicIds));
+        assertTrue(nameCollection.topicNames().containsAll(topicNames));
     }
 
 }

clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java

@@ -122,7 +122,7 @@ private OffsetDeleteResponse buildGroupErrorResponse(Errors error) {
     }
 
     private OffsetDeleteResponse buildPartitionErrorResponse(Errors error) {
-        OffsetDeleteResponse response = new OffsetDeleteResponse(
+        return new OffsetDeleteResponse(
             new OffsetDeleteResponseData()
                 .setThrottleTimeMs(0)
                 .setTopics(new OffsetDeleteResponseTopicCollection(singletonList(
@@ -135,7 +135,6 @@ private OffsetDeleteResponse buildPartitionErrorResponse(Errors error) {
                         ).iterator()))
                 ).iterator()))
         );
-        return response;
     }
 
     private AdminApiHandler.ApiResult<CoordinatorKey, Map<TopicPartition, Errors>> handleWithGroupError(

clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandlerTest.java

@@ -75,13 +75,12 @@ public void testFailedHandleResponse() {
     }
 
     private DeleteGroupsResponse buildResponse(Errors error) {
-        DeleteGroupsResponse response = new DeleteGroupsResponse(
+        return new DeleteGroupsResponse(
                 new DeleteGroupsResponseData()
                     .setResults(new DeletableGroupResultCollection(singletonList(
                             new DeletableGroupResult()
                                 .setErrorCode(error.code())
                                 .setGroupId(groupId1)).iterator())));
-        return response;
     }
 
     private AdminApiHandler.ApiResult<CoordinatorKey, Void> handleWithError(

clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java

@@ -296,7 +296,7 @@ private ConsumerGroupDescribeResponse buildConsumerGroupDescribeResponse(Errors
     }
 
     private DescribeGroupsResponse buildResponse(Errors error, String protocolType) {
-        DescribeGroupsResponse response = new DescribeGroupsResponse(
+        return new DescribeGroupsResponse(
                 new DescribeGroupsResponseData()
                     .setGroups(singletonList(
                             new DescribedGroup()
@@ -314,7 +314,6 @@ private DescribeGroupsResponse buildResponse(Errors error, String protocolType)
                                             .setMemberAssignment(ConsumerProtocol.serializeAssignment(
                                                     new Assignment(new ArrayList<>(tps))).array())
                                             )))));
-        return response;
     }
 
     private AdminApiHandler.ApiResult<CoordinatorKey, ConsumerGroupDescription> handleClassicGroupWithError(

clients/src/test/java/org/apache/kafka/clients/admin/internals/FenceProducersHandlerTest.java

@@ -125,8 +125,6 @@ private ApiResult<CoordinatorKey, ProducerIdAndEpoch> handleResponseError(
         String transactionalId,
         Errors error
     ) {
-        int brokerId = 1;
-
         CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
         Set<CoordinatorKey> keys = mkSet(key);
 

clients/src/test/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandlerTest.java

@@ -90,27 +90,25 @@ public void testFailedHandleResponseInMemberLevel() {
     }
 
     private LeaveGroupResponse buildResponse(Errors error) {
-        LeaveGroupResponse response = new LeaveGroupResponse(
+        return new LeaveGroupResponse(
             new LeaveGroupResponseData()
                 .setErrorCode(error.code())
                 .setMembers(singletonList(
                     new MemberResponse()
                         .setErrorCode(Errors.NONE.code())
                         .setMemberId("m1")
                         .setGroupInstanceId("m1-gii"))));
-        return response;
     }
 
     private LeaveGroupResponse buildResponseWithMemberError(Errors error) {
-        LeaveGroupResponse response = new LeaveGroupResponse(
+        return new LeaveGroupResponse(
             new LeaveGroupResponseData()
                 .setErrorCode(Errors.NONE.code())
                 .setMembers(singletonList(
                     new MemberResponse()
                         .setErrorCode(error.code())
                         .setMemberId("m1")
                         .setGroupInstanceId("m1-gii"))));
-        return response;
     }
 
     private AdminApiHandler.ApiResult<CoordinatorKey, Map<MemberIdentity, Errors>> handleWithGroupError(

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerGroupMetadataTest.java

@@ -28,7 +28,7 @@
 
 public class ConsumerGroupMetadataTest {
 
-    private String groupId = "group";
+    private final String groupId = "group";
 
     @Test
     public void testAssignmentConstructor() {

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -192,12 +192,12 @@ public class KafkaConsumerTest {
     private final String memberId = "memberId";
     private final String leaderId = "leaderId";
     private final Optional<String> groupInstanceId = Optional.of("mock-instance");
-    private Map<String, Uuid> topicIds = Stream.of(
+    private final Map<String, Uuid> topicIds = Stream.of(
             new AbstractMap.SimpleEntry<>(topic, topicId),
             new AbstractMap.SimpleEntry<>(topic2, topicId2),
             new AbstractMap.SimpleEntry<>(topic3, topicId3))
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    private Map<Uuid, String> topicNames = Stream.of(
+    private final Map<Uuid, String> topicNames = Stream.of(
             new AbstractMap.SimpleEntry<>(topicId, topic),
             new AbstractMap.SimpleEntry<>(topicId2, topic2),
             new AbstractMap.SimpleEntry<>(topicId3, topic3))
@@ -1840,28 +1840,28 @@ public void testOperationsBySubscribingConsumerWithDefaultGroupId(GroupProtocol
             // OK, expected
         }
 
-        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, (String) null)) {
+        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
             consumer.subscribe(Collections.singleton(topic));
             fail("Expected an InvalidGroupIdException");
         } catch (InvalidGroupIdException e) {
             // OK, expected
         }
 
-        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, (String) null)) {
+        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
             consumer.committed(Collections.singleton(tp0)).get(tp0);
             fail("Expected an InvalidGroupIdException");
         } catch (InvalidGroupIdException e) {
             // OK, expected
         }
 
-        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, (String) null)) {
+        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
             consumer.commitAsync();
             fail("Expected an InvalidGroupIdException");
         } catch (InvalidGroupIdException e) {
             // OK, expected
         }
 
-        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, (String) null)) {
+        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
             consumer.commitSync();
             fail("Expected an InvalidGroupIdException");
         } catch (InvalidGroupIdException e) {

clients/src/test/java/org/apache/kafka/clients/consumer/RoundRobinAssignorTest.java

@@ -36,12 +36,11 @@
 
 public class RoundRobinAssignorTest {
 
-    private RoundRobinAssignor assignor = new RoundRobinAssignor();
-    private String topic = "topic";
-    private String consumerId = "consumer";
-
-    private String topic1 = "topic1";
-    private String topic2 = "topic2";
+    private final RoundRobinAssignor assignor = new RoundRobinAssignor();
+    private final String topic = "topic";
+    private final String consumerId = "consumer";
+    private final String topic1 = "topic1";
+    private final String topic2 = "topic2";
 
     @Test
     public void testOneConsumerNoTopic() {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java

@@ -1018,7 +1018,7 @@ public void testHeartbeatInstanceFencedResponseWithOldGeneration() throws Interr
     }
 
     @Test
-    public void testHeartbeatRequestWithFencedInstanceIdException() throws InterruptedException {
+    public void testHeartbeatRequestWithFencedInstanceIdException() {
         setupCoordinator();
         mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
 
@@ -1418,11 +1418,7 @@ public void testWakeupAfterJoinGroupReceivedExternalCompletion() throws Exceptio
         mockClient.prepareResponse(syncGroupResponse(Errors.NONE));
         AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
 
-        try {
-            coordinator.ensureActiveGroup();
-            fail("Should have woken up from ensureActiveGroup()");
-        } catch (WakeupException e) {
-        }
+        assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
 
         assertEquals(1, coordinator.onJoinPrepareInvokes);
         assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1458,11 +1454,7 @@ public boolean matches(AbstractRequest body) {
         }, syncGroupResponse(Errors.NONE));
         AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
 
-        try {
-            coordinator.ensureActiveGroup();
-            fail("Should have woken up from ensureActiveGroup()");
-        } catch (WakeupException e) {
-        }
+        assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
 
         assertEquals(1, coordinator.onJoinPrepareInvokes);
         assertEquals(0, coordinator.onJoinCompleteInvokes);
@@ -1526,11 +1518,7 @@ public void testWakeupAfterSyncGroupReceivedExternalCompletion() throws Exceptio
         }, syncGroupResponse(Errors.NONE));
         AtomicBoolean heartbeatReceived = prepareFirstHeartbeat();
 
-        try {
-            coordinator.ensureActiveGroup();
-            fail("Should have woken up from ensureActiveGroup()");
-        } catch (WakeupException e) {
-        }
+        assertThrows(WakeupException.class, () -> coordinator.ensureActiveGroup(), "Should have woken up from ensureActiveGroup()");
 
         assertEquals(1, coordinator.onJoinPrepareInvokes);
         assertEquals(0, coordinator.onJoinCompleteInvokes);

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -142,12 +142,7 @@
 @SuppressWarnings("unchecked")
 public class AsyncKafkaConsumerTest {
 
-    private long retryBackoffMs = 100L;
-    private int defaultApiTimeoutMs = 1000;
-    private boolean autoCommitEnabled = true;
-
     private AsyncKafkaConsumer<String, String> consumer = null;
-
     private Time time = new MockTime(0);
     private final FetchCollector<String, String> fetchCollector = mock(FetchCollector.class);
     private final ApplicationEventHandler applicationEventHandler = mock(ApplicationEventHandler.class);
@@ -208,6 +203,9 @@ private AsyncKafkaConsumer<String, String> newConsumer(
         List<ConsumerPartitionAssignor> assignors,
         String groupId,
         String clientId) {
+        long retryBackoffMs = 100L;
+        int defaultApiTimeoutMs = 1000;
+        boolean autoCommitEnabled = true;
         return new AsyncKafkaConsumer<>(
             new LogContext(),
             clientId,