KAFKA-17554: Flaky testFutureCompletionOutsidePoll in ConsumerNetworkClientTest

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

@@ -41,7 +41,6 @@
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.test.TestUtils;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import java.time.Duration;
@@ -266,14 +265,11 @@ public void testMetadataFailurePropagated() {
         assertEquals(metadataException, exc);
     }
 
-    @Disabled("KAFKA-17554")
     @Test
     public void testFutureCompletionOutsidePoll() throws Exception {
         // Tests the scenario in which the request that is being awaited in one thread
         // is received and completed in another thread.
-
         final CountDownLatch t1TheardCountDownLatch = new CountDownLatch(1);
-        final CountDownLatch t2ThreadCountDownLatch = new CountDownLatch(2);
 
         final RequestFuture<ClientResponse> future = consumerClient.send(node, heartbeat());
         consumerClient.pollNoWakeup(); // dequeue and send the request
@@ -282,33 +278,19 @@ public void testFutureCompletionOutsidePoll() throws Exception {
         Thread t1 = new Thread(() ->  {
             t1TheardCountDownLatch.countDown();
             consumerClient.pollNoWakeup();
-            t2ThreadCountDownLatch.countDown();
         });
 
         t1.start();
-
-        Thread t2 = new Thread(() -> {
-            try {
-                t2ThreadCountDownLatch.await();
-                consumerClient.poll(future);
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-        });
-        t2.start();
 
         // Simulate a network response and return from the poll in t1
         client.respond(heartbeatResponse(Errors.NONE));
         // Wait for t1 to block in poll
         t1TheardCountDownLatch.await();
-
         client.wakeup();
-        // while t1 is blocked in poll, t2 should be able to complete the future
-        t2ThreadCountDownLatch.countDown();
-
-        // Both threads should complete since t1 should wakeup t2
+
         t1.join();
-        t2.join();
+        consumerClient.poll(future);
+
         assertTrue(future.succeeded());
     }