NettyHttpClient: Fix double-return on certain exceptions. (#12626)

The "exceptionCaught" handler may get called multiple times. We should
only return the channel to the pool the first time. Returning it more
than once leads to a warning like "Resource at key[%s] was returned
multiple times?"

core/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java

@@ -317,6 +317,8 @@ public class NettyHttpClient extends AbstractHttpClient
               }
             }
 
+            // Ignore return value of setException, since exceptionCaught can be called multiple times and we
+            // only want to report the first one.
             if (event.getCause() instanceof ReadTimeoutException) {
               // ReadTimeoutException thrown by ReadTimeoutHandler is a singleton with a misleading stack trace.
               // No point including it: instead, we replace it with a fresh exception.
@@ -338,7 +340,11 @@ public class NettyHttpClient extends AbstractHttpClient
               log.warn(e, "Error while closing channel");
             }
             finally {
-              channelResourceContainer.returnResource();
+              if (channelResourceContainer.isPresent()) {
+                // exceptionCaught can be called multiple times: we only want to return the channel if it hasn't
+                // already been returned.
+                channelResourceContainer.returnResource();
+              }
             }
           }
 

core/src/main/java/org/apache/druid/java/util/http/client/pool/ResourceContainer.java

@@ -24,5 +24,6 @@ package org.apache.druid.java.util.http.client.pool;
 public interface ResourceContainer<ResourceType>
 {
   ResourceType get();
+  boolean isPresent();
   void returnResource();
 }

core/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java

@@ -119,11 +119,17 @@ public class ResourcePool<K, V> implements Closeable
         return value;
       }
 
+      @Override
+      public boolean isPresent()
+      {
+        return !returned.get();
+      }
+
       @Override
       public void returnResource()
       {
         if (returned.getAndSet(true)) {
-          log.warn(StringUtils.format("Resource at key[%s] was returned multiple times?", key));
+          log.warn("Resource at key[%s] was returned multiple times?", key);
         } else {
           holder.giveBack(value);
         }