Bump async-http-client to 3.0.1

Bump async-http-client to 3.0.1 to resolve CVE-2024-53990 _.

pom.xml

@@ -1010,8 +1010,7 @@
             <dependency>
                 <groupId>org.asynchttpclient</groupId>
                 <artifactId>async-http-client</artifactId>
-                <!-- Uses Netty 4.1.x -->
-                <version>2.5.3</version>
+                <version>3.0.1</version>
             </dependency>
             <dependency>
                 <groupId>net.java.dev.jna</groupId>

processing/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java

@@ -23,7 +23,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
-import com.google.common.primitives.Ints;
 import io.netty.handler.codec.http.HttpHeaders;
 import org.apache.druid.concurrent.ConcurrentAwaitableCounter;
 import org.apache.druid.java.util.common.ISE;
@@ -45,6 +44,7 @@
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayDeque;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -773,7 +773,7 @@ private void send(byte[] buffer, int length) throws Exception
         request.setHeader(HttpHeaders.Names.AUTHORIZATION, "Basic " + encoded);
       }
 
-      request.setRequestTimeout(Ints.saturatedCast(timeoutMillis));
+      request.setRequestTimeout(Duration.ofMillis(timeoutMillis));
 
       ListenableFuture<Response> future = client.executeRequest(request);
       Response response;

processing/src/test/java/org/apache/druid/java/util/emitter/core/HttpEmitterTest.java

@@ -30,6 +30,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class HttpEmitterTest
@@ -56,8 +57,8 @@ public void setup()
       @Override
       protected ListenableFuture<Response> go(Request request)
       {
-        int timeout = request.getRequestTimeout();
-        timeoutUsed.set(timeout);
+        Duration timeout = request.getRequestTimeout();
+        timeoutUsed.set(timeout.toMillis());
         return GoHandlers.immediateFuture(EmitterTest.okResponse());
       }
     });

processing/src/test/java/org/apache/druid/java/util/http/client/AsyncHttpClientTest.java

@@ -30,6 +30,7 @@
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -81,7 +82,7 @@ public void run()
       requestStart = System.currentTimeMillis();
       Future<?> future = client
           .prepareGet(StringUtils.format("http://localhost:%d/", serverSocket.getLocalPort()))
-          .setRequestTimeout(2000)
+          .setRequestTimeout(Duration.ofMillis(2000))
           .execute();
       System.out.println("created future in: " + (System.currentTimeMillis() - requestStart));
       future.get(3000, TimeUnit.MILLISECONDS);
@@ -103,7 +104,7 @@ private void warmUp(ServerSocket serverSocket, DefaultAsyncHttpClient client)
     try {
       Future<?> future = client
           .prepareGet(StringUtils.format("http://localhost:%d/", serverSocket.getLocalPort()))
-          .setRequestTimeout(100)
+          .setRequestTimeout(Duration.ofMillis(100))
           .execute();
       future.get();
     }