KAFKA-7513: Fix timing issue in SaslAuthenticatorFailureDelayTest (#5805

)

Reduce tick interval of the mock timer and avoid large timer increments to avoid hitting idle expiry on the client-side before delayed close is processed by the server. Also reduce poll interval in the server to make the test complete faster (since delayed close is only processed when poll returns).

Reviewers: Ismael Juma <ismael@juma.me.uk>

clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java

@@ -28,7 +28,6 @@
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.security.authenticator.CredentialCache;
-import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.test.TestUtils;
@@ -87,7 +86,7 @@ public static void waitForChannelReady(Selector selector, String node) throws IO
         assertTrue(selector.isChannelReady(node));
     }
 
-    public static ChannelState waitForChannelClose(Selector selector, String node, ChannelState.State channelState, MockTime mockTime)
+    public static ChannelState waitForChannelClose(Selector selector, String node, ChannelState.State channelState)
             throws IOException {
         boolean closed = false;
         for (int i = 0; i < 300; i++) {
@@ -96,19 +95,13 @@ public static ChannelState waitForChannelClose(Selector selector, String node, C
                 closed = true;
                 break;
             }
-            if (mockTime != null)
-                mockTime.setCurrentTimeMs(mockTime.milliseconds() + 150);
         }
         assertTrue("Channel was not closed by timeout", closed);
         ChannelState finalState = selector.disconnected().get(node);
         assertEquals(channelState, finalState.state());
         return finalState;
     }
 
-    public static ChannelState waitForChannelClose(Selector selector, String node, ChannelState.State channelState) throws IOException {
-        return waitForChannelClose(selector, node, channelState, null);
-    }
-
     public static void completeDelayedChannelClose(Selector selector, long currentTimeNanos) {
         selector.completeDelayedChannelClose(currentTimeNanos);
     }

clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java

@@ -153,7 +153,7 @@ public void run() {
         try {
             acceptorThread.start();
             while (serverSocketChannel.isOpen()) {
-                selector.poll(1000);
+                selector.poll(100);
                 synchronized (newChannels) {
                     for (SocketChannel socketChannel : newChannels) {
                         String id = id(socketChannel);

clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java

@@ -53,8 +53,8 @@
 @RunWith(value = Parameterized.class)
 public class SaslAuthenticatorFailureDelayTest {
     private static final int BUFFER_SIZE = 4 * 1024;
-    private static MockTime time = new MockTime(50);
 
+    private final MockTime time = new MockTime(10);
     private NioEchoServer server;
     private Selector selector;
     private ChannelBuilder channelBuilder;
@@ -221,7 +221,7 @@ private ChannelState createAndCheckClientConnectionFailure(SecurityProtocol secu
             throws Exception {
         createClientConnection(securityProtocol, node);
         ChannelState finalState = NetworkTestUtils.waitForChannelClose(selector, node,
-                ChannelState.State.AUTHENTICATION_FAILED, time);
+                ChannelState.State.AUTHENTICATION_FAILED);
         selector.close();
         selector = null;
         return finalState;