/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.client.impl;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import io.netty.channel.ConnectTimeoutException;
import org.apache.pulsar.client.api.PulsarClient;
import org.testng.Assert;
import org.testng.annotations.Test;

public class ConnectionTimeoutTest {

    // 192.0.2.0/24 is assigned for documentation, should be a deadend
    static final String blackholeBroker = "pulsar://192.0.2.1:1234";

    @Test
    public void testLowTimeout() throws Exception {
        try (PulsarClient clientLow = PulsarClient.builder().serviceUrl(blackholeBroker)
                .connectionTimeout(1, TimeUnit.MILLISECONDS)
                .operationTimeout(1000, TimeUnit.MILLISECONDS).build()) {
            CompletableFuture<?> lowFuture = clientLow.newProducer().topic("foo").createAsync();

            try {
                lowFuture.get();
                Assert.fail("Shouldn't be able to connect to anything");
            } catch (Exception e) {
                Assert.assertEquals(e.getCause().getCause().getCause().getClass(), ConnectTimeoutException.class);
            }
        }
    }
}

@@ -842,16 +842,227 @@
 .impl;%0A%0A
+import io.netty.channel.ConnectTimeoutException;%0Aimport java.io.IOException;%0Aimport java.net.InetAddress;%0Aimport java.net.ServerSocket;%0Aimport java.net.Socket;%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0A
 import j
@@ -1100,16 +1100,60 @@
 Future;%0A
+import java.util.concurrent.CountDownLatch;%0A
 import j
@@ -1193,32 +1193,29 @@
 ort 
-io.netty.channel.Connect
+java.util.concurrent.
 Time
@@ -1387,207 +1387,1377 @@
     
-// 192.0.2.0/24 is assigned for documentation, should be a deadend%0A    static final String blackholeBroker = %22pulsar://192.0.2.1:1234%22;%0A%0A    @Test%0A    public void testLowTimeout() throws Exception %7B%0A
+@Test%0A    public void testLowTimeout() throws Exception %7B%0A        int backlogSize = 1;%0A        // create a dummy server and fill the backlog of the server so that it won't respond%0A        // so that the client timeout can be tested with this server%0A        try (ServerSocket serverSocket = new ServerSocket(0, backlogSize, InetAddress.getByName(%22localhost%22))) %7B%0A            CountDownLatch latch = new CountDownLatch(backlogSize + 1);%0A            List%3CThread%3E threads = new ArrayList%3C%3E();%0A            for (int i = 0; i %3C backlogSize + 1; i++) %7B%0A                Thread connectThread = new Thread(() -%3E %7B%0A                    try (Socket socket = new Socket()) %7B%0A                        socket.connect(serverSocket.getLocalSocketAddress());%0A                        latch.countDown();%0A                        Thread.sleep(10000L);%0A                    %7D catch (IOException e) %7B%0A                        throw new RuntimeException(e);%0A                    %7D catch (InterruptedException e) %7B%0A                        Thread.currentThread().interrupt();%0A                    %7D%0A                %7D);%0A                connectThread.start();%0A                threads.add(connectThread);%0A            %7D%0A            latch.await();%0A%0A            String blackholeBroker =%0A                    %22pulsar://%22 + serverSocket.getInetAddress().getHostAddress() + %22:%22 + serverSocket.getLocalPort();%0A%0A    
     
@@ -2857,16 +2857,20 @@
         
+    
 .connect
@@ -2906,16 +2906,20 @@
 ECONDS)%0A
+    
         
@@ -2996,16 +2996,20 @@
         
+    
 Completa
@@ -3081,25 +3081,28 @@
 ateAsync();%0A
-%0A
+    
             
@@ -3123,16 +3123,20 @@
         
+    
 lowFutur
@@ -3141,19 +3141,43 @@
 ure.get(
-);%0A
+10, TimeUnit.SECONDS);%0A    
         
@@ -3245,32 +3245,36 @@
 %22);%0A            
+    
 %7D catch (Excepti
@@ -3266,16 +3266,23 @@
  catch (
+Timeout
 Exceptio
@@ -3308,104 +3308,320 @@
     
-Assert.assertEquals(e.getCause().getCause().getCause().getClass(), ConnectTimeoutException.class
+    Assert.fail(%22Connection timeout didn't apply.%22);%0A                %7D catch (Exception e) %7B%0A                    Assert.assertEquals(e.getCause().getCause().getCause().getClass(), ConnectTimeoutException.class);%0A                %7D%0A            %7D finally %7B%0A                threads.stream().forEach(Thread::interrupt
 );%0A 

Fix flaky test ConnectionTimeoutTest (#15865)