MINOR: Cleanup in ShareConsumerTest (#20940)

*What*
PR fixes some typos/nit in `ShareConsumerTest` and fixes a test
assertion in `ConsumerNetworkThreadTest`.

Reviewers: Andrew Schofield <[email protected]>, Chia-Ping Tsai
 <[email protected]>

Author: ShivsundarR

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

@@ -3163,7 +3163,7 @@ public void testDescribeShareGroupOffsetsForEmptySharePartition() {
         try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(groupId);
              Admin adminClient = createAdminClient()) {
             shareConsumer.subscribe(List.of(tp.topic()));
-            // Polling share consumer to make sure the share partition in created.
+            // Polling share consumer to make sure the share partition is created.
             shareConsumer.poll(Duration.ofMillis(2000));
             SharePartitionOffsetInfo sharePartitionOffsetInfo = sharePartitionOffsetInfo(adminClient, groupId, tp);
             // Since the partition is empty, and no records have been consumed, the share partition startOffset will be
@@ -3185,7 +3185,7 @@ public void testSharePartitionLagForSingleShareConsumer() {
             producer.send(record);
             producer.flush();
             shareConsumer.subscribe(List.of(tp.topic()));
-            // Polling share consumer to make sure the share partition in created and teh record is consumed.
+            // Polling share consumer to make sure the share partition is created and the record is consumed.
             waitedPoll(shareConsumer, 2500L, 1);
             // Acknowledge and commit the consumed record to update the share partition state.
             shareConsumer.commitSync();
@@ -3212,10 +3212,9 @@ public void testSharePartitionLagForMultipleShareConsumers() {
             ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "Message".getBytes());
             producer.send(record);
             producer.flush();
-            producer.flush();
             shareConsumer1.subscribe(List.of(tp.topic()));
             shareConsumer2.subscribe(List.of(tp.topic()));
-            // Polling share consumer 1 to make sure the share partition in created and the records are consumed.
+            // Polling share consumer 1 to make sure the share partition is created and the records are consumed.
             waitedPoll(shareConsumer1, 2500L, 1);
             // Acknowledge and commit the consumed records to update the share partition state.
             shareConsumer1.commitSync();
@@ -3769,7 +3768,6 @@ private SharePartitionOffsetInfo sharePartitionOffsetInfo(Admin adminClient, Str
     private void verifySharePartitionLag(Admin adminClient, String groupId, TopicPartition tp, long expectedLag) throws InterruptedException {
         TestUtils.waitForCondition(() -> {
             SharePartitionOffsetInfo sharePartitionOffsetInfo = sharePartitionOffsetInfo(adminClient, groupId, tp);
-            System.out.println("Current share partition description: " + sharePartitionOffsetInfo);
             return sharePartitionOffsetInfo != null &&
                 sharePartitionOffsetInfo.lag().isPresent() &&
                 sharePartitionOffsetInfo.lag().get() == expectedLag;

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

@@ -35,7 +35,6 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
-import org.mockito.ArgumentMatchers;
 
 import java.time.Duration;
 import java.util.LinkedList;
@@ -52,6 +51,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -204,7 +204,7 @@ public void testRunOnceInvokesReaper() {
     public void testSendUnsentRequests() {
         when(networkClientDelegate.hasAnyPendingRequests()).thenReturn(true).thenReturn(true).thenReturn(false);
         consumerNetworkThread.cleanup();
-        verify(networkClientDelegate, times(2)).poll(anyLong(), anyLong(), ArgumentMatchers.booleanThat(onClose -> onClose));
+        verify(networkClientDelegate, times(2)).poll(anyLong(), anyLong(), eq(true));
     }
 
     @ParameterizedTest