This is an automated email from the ASF dual-hosted git repository.
kgyrtkirk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new d2bff3c9ec5 feat: add adaptive http pooling (#20273)
d2bff3c9ec5 is described below
commit d2bff3c9ec541e0ce5b6e4fcdb9b79a06811cc3b
Author: Zoltan Haindrich <[email protected]>
AuthorDate: Mon Sep 28 11:03:09 2026 +0200
feat: add adaptive http pooling (#20273)
---
docs/configuration/index.md | 3 +
docs/operations/metrics.md | 20 +
.../java/util/http/client/HttpClientConfig.java | 17 +
.../java/util/http/client/HttpClientInit.java | 8 +-
.../java/util/http/client/NettyHttpClient.java | 5 +
.../java/util/http/client/pool/ResourcePool.java | 536 +++++++++++-
.../util/http/client/pool/ResourcePoolConfig.java | 16 +
.../util/http/client/pool/ResourcePoolTest.java | 901 +++++++++++++--------
.../druid/guice/http/DruidHttpClientConfig.java | 11 +-
.../apache/druid/guice/http/HttpClientModule.java | 22 +-
.../server/metrics/HttpClientPoolMonitor.java | 91 +++
.../server/metrics/HttpClientPoolRegistry.java | 56 ++
.../guice/http/DruidHttpClientConfigTest.java | 90 ++
.../guice/http/HttpClientPoolRegistrationTest.java | 69 ++
.../server/metrics/HttpClientPoolMonitorTest.java | 202 +++++
15 files changed, 1657 insertions(+), 390 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 3e02f3b6d23..4eff8506d41 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -689,6 +689,7 @@ All Druid components can communicate with each other over
HTTP.
|`druid.global.http.clientConnectTimeout`|Connect timeout (in milliseconds)
for the HTTP client used to forward management API requests between Druid
services. On the Router, this covers forwarding management API calls to the
Coordinator or Overlord. On the Coordinator, this covers proxying
`/druid/indexer/*` requests to the Overlord (when they run as separate
processes). Does not affect Router query proxying to Brokers (see
`druid.router.http.clientConnectTimeout`) or direct RPC connect [...]
|`druid.global.http.connectTimeout`|Connect timeout for the HTTP client used
for most direct RPC between Druid services. This covers, among other things,
Overlord-to-task and supervisor-to-task calls in the indexing service,
Coordinator lookup management, dynamic config sync between services, MSQ tasks
reading from data servers, and general Coordinator/Overlord/Broker service
clients. Does not affect Broker-to-Historical query dispatch (see
`druid.broker.http.connectTimeout`) or request [...]
|`druid.global.http.allocator`|Netty memory allocator used by the direct-RPC
HTTP client. Accepts `adaptive` (adaptive between `pooled` and `unpooled` based
on load), `pooled`, or `unpooled`.|`adaptive`|
+|`druid.global.http.poolImplementation`|How the connection pool tracks demand,
never exceeding `numConnections` either way. With `adaptive`, a request
discards every stale or broken connection it walks past and opens a new one
only once none is left, so the pool falls back to the number of connections the
traffic actually needs. With `retaining`, the pool holds on to every connection
it has opened, replacing a stale or broken one by a fresh one, one for one, so
it stays at its high-water [...]
### Common endpoints configuration
@@ -1834,6 +1835,7 @@ client has the following configuration options.
|`druid.broker.http.numMaxThreads`|`Maximum number of I/O worker
threads|(number of cores) * 3 / 2 + 1`|
|`druid.broker.http.connectTimeout`|Connect timeout for the HTTP client the
Broker uses to dispatch queries to Historical and real-time processes.|`PT10S`|
|`druid.broker.http.allocator`|Netty memory allocator used by the direct-RPC
HTTP client. Accepts `adaptive` (adaptive between `pooled` and `unpooled` based
on load), `pooled`, or `unpooled`.|`adaptive`|
+|`druid.broker.http.poolImplementation`|How the connection pool tracks demand,
never exceeding `numConnections` either way. With `adaptive`, a query discards
every stale or broken connection it walks past and opens a new one only once
none is left, so the pool falls back to the number of connections the traffic
actually needs. With `retaining`, the pool holds on to every connection it has
opened, replacing a stale or broken one by a fresh one, one for one, so it
stays at its high-water m [...]
##### Retry policy
@@ -2008,6 +2010,7 @@ The following table lists available monitors and the
respective services where t
|`org.apache.druid.server.metrics.WorkerTaskCountStatsMonitor`|Reports how
many ingestion tasks are currently running/pending/waiting, the number of
successful/failed tasks, and metrics about task slot usage for the reporting
worker, per emission period. |MiddleManager, Indexer|
|`org.apache.druid.server.metrics.ServiceStatusMonitor`|Reports a heartbeat
for the service.|Any|
|`org.apache.druid.server.metrics.GroupByStatsMonitor`|Report metrics for
groupBy queries like disk and merge buffer utilization. |Broker, Historical,
Indexer, Peon|
+|`org.apache.druid.server.metrics.HttpClientPoolMonitor`|Reports connection
churn and usage of the HTTP client connection pools used for service to service
communication, per remote end, per emission period.|Any|
For example, if you only wanted monitors on all services for system and JVM
information, you'd add the following to `common.runtime.properties`:
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index c30d34d14dc..0700f6db464 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -553,6 +553,26 @@ These metrics are emitted when
`druid.auth.emitAuthMetrics` is set to `true`.
|`segment/rowCount/avg`| The average number of rows per segment on a
historical. `SegmentStatsMonitor` must be enabled.| `dataSource`, `tier`,
`priority`|Varies. See [segment
optimization](../operations/segment-optimization.md) for guidance on optimal
segment sizes. |
|`segment/rowCount/range/count`| The number of segments in a bucket.
`SegmentStatsMonitor` must be enabled.| `dataSource`, `tier`, `priority`,
`range`|Varies|
+### HTTP client connection pools
+
+These metrics are only available if the `HttpClientPoolMonitor` module is
included in `druid.monitoring.monitors`.
+They cover the connection pools that Druid services use to talk to each other,
one emission per remote end. The
+`server` dimension is that remote end, and the `httpClient` dimension names
the client that pools the connections to
+it: `client` and `escalatedClient` are configured by `druid.broker.http`,
`global` and `escalatedGlobal` by
+`druid.global.http`. A remote end that a service has stopped talking to keeps
being reported, with zeroes, until the
+service is restarted.
+
+|Metric|Description|Dimensions|Normal value|
+|------|-----------|----------|------------|
+|`httpClient/pool/opened`|Number of connections opened.|`httpClient`,
`server`|Varies. Steady churn on an idle cluster points at connections being
discarded too eagerly.|
+|`httpClient/pool/closed`|Number of connections closed, whether they were
broken, unused for too long, or surplus.|`httpClient`, `server`|Varies|
+|`httpClient/pool/errored`|Number of failures while opening, health checking,
or closing a connection.|`httpClient`, `server`|0|
+|`httpClient/pool/timedOut`|Number of connections discarded for being unused
longer than `unusedConnectionTimeout`.|`httpClient`, `server`|Varies|
+|`httpClient/pool/taken`|Number of connections handed to a caller, that is,
the number of requests that got a connection.|`httpClient`, `server`|Varies|
+|`httpClient/pool/returned`|Number of connections given back by a
caller.|`httpClient`, `server`|Close to `httpClient/pool/taken`|
+|`httpClient/pool/used`|Number of connections in the hands of callers at the
time of the emission, that is, the requests in flight to that remote end. A
level, not a per period count.|`httpClient`, `server`|≤
`druid.<service>.http.numConnections`. Sitting at that ceiling means requests
are waiting for a connection.|
+|`httpClient/pool/idle`|Number of connections parked for the next caller at
the time of the emission. A level, not a per period count.|`httpClient`,
`server`|≤ `druid.<service>.http.numConnections`|
+
### JVM
These metrics are only available if the `JvmMonitor` module is included in
`druid.monitoring.monitors`.
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientConfig.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientConfig.java
index a1dfe5cdaf4..de650ff92af 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientConfig.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientConfig.java
@@ -21,6 +21,7 @@ package org.apache.druid.java.util.http.client;
import io.netty.buffer.AdaptiveByteBufAllocator;
import io.netty.buffer.ByteBufAllocator;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
import org.apache.druid.utils.JvmUtils;
import org.joda.time.Duration;
import org.joda.time.Period;
@@ -89,6 +90,7 @@ public class HttpClientConfig
private final int numConnections;
private final boolean eagerInitialization;
+ private final ResourcePool.Implementation poolImplementation;
private final SSLContext sslContext;
private final HttpClientProxyConfig proxyConfig;
private final Duration readTimeout;
@@ -102,6 +104,7 @@ public class HttpClientConfig
private HttpClientConfig(
int numConnections,
boolean eagerInitialization,
+ ResourcePool.Implementation poolImplementation,
SSLContext sslContext,
HttpClientProxyConfig proxyConfig,
Duration readTimeout,
@@ -115,6 +118,7 @@ public class HttpClientConfig
{
this.numConnections = numConnections;
this.eagerInitialization = eagerInitialization;
+ this.poolImplementation = poolImplementation;
this.sslContext = sslContext;
this.proxyConfig = proxyConfig;
this.readTimeout = readTimeout;
@@ -136,6 +140,11 @@ public class HttpClientConfig
return eagerInitialization;
}
+ public ResourcePool.Implementation getPoolImplementation()
+ {
+ return poolImplementation;
+ }
+
public SSLContext getSslContext()
{
return sslContext;
@@ -185,6 +194,7 @@ public class HttpClientConfig
{
private int numConnections = 1;
private boolean eagerInitialization = true;
+ private ResourcePool.Implementation poolImplementation =
ResourcePool.Implementation.ADAPTIVE;
private SSLContext sslContext = null;
private HttpClientProxyConfig proxyConfig = null;
private Duration readTimeout = null;
@@ -211,6 +221,12 @@ public class HttpClientConfig
return this;
}
+ public Builder withPoolImplementation(ResourcePool.Implementation
poolImplementation)
+ {
+ this.poolImplementation = poolImplementation;
+ return this;
+ }
+
public Builder withSslContext(SSLContext sslContext)
{
this.sslContext = sslContext;
@@ -281,6 +297,7 @@ public class HttpClientConfig
return new HttpClientConfig(
numConnections,
eagerInitialization,
+ poolImplementation,
sslContext,
proxyConfig,
readTimeout,
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java
index b57d53bd1cf..3c3c562b5d9 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java
@@ -47,6 +47,11 @@ import java.util.concurrent.TimeUnit;
public class HttpClientInit
{
public static HttpClient createClient(HttpClientConfig config, Lifecycle
lifecycle)
+ {
+ return createNettyClient(config, lifecycle);
+ }
+
+ public static NettyHttpClient createNettyClient(HttpClientConfig config,
Lifecycle lifecycle)
{
try {
return lifecycle.addMaybeStartManagedInstance(
@@ -65,7 +70,8 @@ public class HttpClientInit
),
new ResourcePoolConfig(
config.getNumConnections(),
- config.getUnusedConnectionTimeoutDuration().getMillis()
+ config.getUnusedConnectionTimeoutDuration().getMillis(),
+ config.getPoolImplementation()
),
config.isEagerInitialization()
),
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
index 4c712666761..ff4f64af394 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java
@@ -87,6 +87,11 @@ public class NettyHttpClient extends AbstractHttpClient
this.compressionCodec = Preconditions.checkNotNull(compressionCodec);
}
+ public ResourcePool<String, ChannelFuture> getPool()
+ {
+ return pool;
+ }
+
@LifecycleStart
public void start()
{
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java
index dfa9903b900..f1a6e493878 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePool.java
@@ -19,11 +19,14 @@
package org.apache.druid.java.util.http.client.pool;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.RateLimiter;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.common.logger.Logger;
@@ -32,31 +35,31 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
/**
- * A resource pool based on {@link LoadingCache}. When a resource is first
requested for a new key,
- * If the flag: eagerInitialization is true: use {@link
EagerCreationResourceHolder}
- * {@link ResourcePoolConfig#getMaxPerKey()} resources are initialized and
cached in the {@link #pool}.
- * Else:
- * Initialize a single resource and further lazily using {@link
LazyCreationResourceHolder}
- * The individual resource in {@link ResourceHolderPerKey} is valid while
(current time - last access time)
- * <= {@link ResourcePoolConfig#getUnusedConnectionTimeoutMillis()}.
+ * Lends out at most {@link ResourcePoolConfig#getMaxPerKey()} resources per
key, blocking further takers until one
+ * comes back.
*
- * A resource is closed and reinitialized if {@link ResourceFactory#isGood}
returns false or it's expired based on
- * {@link ResourcePoolConfig#getUnusedConnectionTimeoutMillis()}.
+ * A resource is discarded once it has gone {@link
ResourcePoolConfig#getUnusedConnectionTimeoutMillis()} unused or
+ * {@link ResourceFactory#isGood} rejects it. With eagerInitialization a key
starts out full, otherwise empty.
*
- * {@link ResourcePoolConfig#getMaxPerKey() is a hard limit for the max number
of resources per cache entry. The total
- * number of resources in {@link ResourceHolderPerKey} cannot be larger than
the limit in any case.
+ * {@link ResourcePoolConfig#getPoolImplementation()} selects which {@link
Implementation} does the pooling.
*/
public class ResourcePool<K, V> implements Closeable
{
private static final Logger log = new Logger(ResourcePool.class);
- private final LoadingCache<K, ResourceHolderPerKey<K, V>> pool;
+ private final LoadingCache<K, PooledResources<V>> pool;
private final AtomicBoolean closed = new AtomicBoolean(false);
public ResourcePool(final ResourceFactory<K, V> factory, final
ResourcePoolConfig config,
@@ -66,28 +69,27 @@ public class ResourcePool<K, V> implements Closeable
new CacheLoader<>()
{
@Override
- public ResourceHolderPerKey<K, V> load(K input)
+ public PooledResources<V> load(K input)
{
- if (eagerInitialization) {
- return new EagerCreationResourceHolder<>(
- config.getMaxPerKey(),
- config.getUnusedConnectionTimeoutMillis(),
- input,
- factory
- );
- } else {
- return new LazyCreationResourceHolder<>(
- config.getMaxPerKey(),
- config.getUnusedConnectionTimeoutMillis(),
- input,
- factory
- );
- }
+ return config.getPoolImplementation().create(config, input,
factory, eagerInitialization);
}
}
);
}
+ /**
+ * The {@link Stats} of every key this pool has resources for, resetting the
counted events as it reads them, so
+ * that each call reports what happened since the previous one. Meant for a
single caller.
+ */
+ public Map<K, Stats> drainStats()
+ {
+ final Map<K, Stats> stats = new HashMap<>();
+ for (Map.Entry<K, PooledResources<V>> e : pool.asMap().entrySet()) {
+ stats.put(e.getKey(), e.getValue().drainStats());
+ }
+ return stats;
+ }
+
/**
* Returns a {@link ResourceContainer} for the given key or null if this
pool is already closed.
*/
@@ -99,7 +101,7 @@ public class ResourcePool<K, V> implements Closeable
return null;
}
- final ResourceHolderPerKey<K, V> holder;
+ final PooledResources<V> holder;
try {
holder = pool.get(key);
}
@@ -107,6 +109,10 @@ public class ResourcePool<K, V> implements Closeable
throw new RuntimeException(e);
}
final V value = holder.get();
+ if (value == null) {
+ return null;
+ }
+ holder.counters.taken.incrementAndGet();
return new ResourceContainer<>()
{
@@ -125,6 +131,7 @@ public class ResourcePool<K, V> implements Closeable
if (returned.getAndSet(true)) {
log.warn("Resource at key[%s] was returned multiple times?", key);
} else {
+ holder.counters.returned.incrementAndGet();
holder.giveBack(value);
}
}
@@ -151,11 +158,11 @@ public class ResourcePool<K, V> implements Closeable
public void close()
{
closed.set(true);
- final ConcurrentMap<K, ResourceHolderPerKey<K, V>> mapView = pool.asMap();
+ final ConcurrentMap<K, PooledResources<V>> mapView = pool.asMap();
Closer closer = Closer.create();
- for (Iterator<Map.Entry<K, ResourceHolderPerKey<K, V>>> iterator =
+ for (Iterator<Map.Entry<K, PooledResources<V>>> iterator =
mapView.entrySet().iterator(); iterator.hasNext(); ) {
- Map.Entry<K, ResourceHolderPerKey<K, V>> e = iterator.next();
+ Map.Entry<K, PooledResources<V>> e = iterator.next();
iterator.remove();
closer.register(e.getValue());
}
@@ -167,6 +174,239 @@ public class ResourcePool<K, V> implements Closeable
}
}
+ /**
+ * What happened to the resources of one key since the previous {@link
#drainStats()}, and what the key holds right
+ * now.
+ *
+ * @param errored calls into the {@link ResourceFactory} that threw
+ * @param timedOut resources discarded for going {@link
ResourcePoolConfig#getUnusedConnectionTimeoutMillis()} unused
+ * @param taken resources handed to a caller by {@link #take}
+ * @param returned resources given back through {@link
ResourceContainer#returnResource()}
+ * @param used resources in the hands of callers at the time of the call
+ * @param idle resources waiting for the next caller at the time of the
call
+ */
+ public record Stats(
+ long opened,
+ long closed,
+ long errored,
+ long timedOut,
+ long taken,
+ long returned,
+ int used,
+ int idle
+ )
+ {
+ }
+
+ /**
+ * The counted events of one key, drained on read.
+ */
+ private static class Counters
+ {
+ private final AtomicLong opened = new AtomicLong();
+ private final AtomicLong closed = new AtomicLong();
+ private final AtomicLong errored = new AtomicLong();
+ private final AtomicLong timedOut = new AtomicLong();
+ private final AtomicLong taken = new AtomicLong();
+ private final AtomicLong returned = new AtomicLong();
+
+ private Stats drain(int used, int idle)
+ {
+ return new Stats(
+ opened.getAndSet(0),
+ closed.getAndSet(0),
+ errored.getAndSet(0),
+ timedOut.getAndSet(0),
+ taken.getAndSet(0),
+ returned.getAndSet(0),
+ used,
+ idle
+ );
+ }
+ }
+
+ /**
+ * Records on {@link Counters} what the {@link ResourceFactory} it wraps was
asked to do, so that every
+ * {@link Implementation} is counted the same way.
+ */
+ private static class CountingResourceFactory<K, V> implements
ResourceFactory<K, V>
+ {
+ private final ResourceFactory<K, V> delegate;
+ private final Counters counters;
+
+ private CountingResourceFactory(ResourceFactory<K, V> delegate, Counters
counters)
+ {
+ this.delegate = delegate;
+ this.counters = counters;
+ }
+
+ @Override
+ public V generate(K key)
+ {
+ final V resource;
+ try {
+ resource = delegate.generate(key);
+ }
+ catch (Throwable t) {
+ counters.errored.incrementAndGet();
+ throw t;
+ }
+ if (resource != null) {
+ counters.opened.incrementAndGet();
+ }
+ return resource;
+ }
+
+ @Override
+ public boolean isGood(V resource)
+ {
+ try {
+ return delegate.isGood(resource);
+ }
+ catch (Throwable t) {
+ counters.errored.incrementAndGet();
+ throw t;
+ }
+ }
+
+ @Override
+ public void close(V resource)
+ {
+ try {
+ delegate.close(resource);
+ }
+ catch (Throwable t) {
+ counters.errored.incrementAndGet();
+ throw t;
+ }
+ counters.closed.incrementAndGet();
+ }
+ }
+
+ /**
+ * Which implementation pools the resources of a key.
+ */
+ public enum Implementation
+ {
+ /**
+ * Adaptively follows demand and maximizes active pool size.
+ */
+ ADAPTIVE {
+ @Override
+ <K, V> PooledResources<V> create(
+ ResourcePoolConfig config,
+ K key,
+ ResourceFactory<K, V> factory,
+ boolean eagerInitialization
+ )
+ {
+ final AdaptiveResourceHolderPerKey<K, V> resources = new
AdaptiveResourceHolderPerKey<>(
+ config.getMaxPerKey(),
+ config.getUnusedConnectionTimeoutMillis(),
+ key,
+ factory
+ );
+ if (eagerInitialization) {
+ resources.preload();
+ }
+ return resources;
+ }
+ },
+
+ /**
+ * Never gives up a resource it has opened: a stale or broken one is
replaced by a fresh one, one for one, and only
+ * once it has reached the front of the queue, so the pool stays at its
high-water mark. Guards the resources of a
+ * key with its monitor.
+ */
+ RETAINING {
+ @Override
+ <K, V> PooledResources<V> create(
+ ResourcePoolConfig config,
+ K key,
+ ResourceFactory<K, V> factory,
+ boolean eagerInitialization
+ )
+ {
+ if (eagerInitialization) {
+ return new EagerCreationResourceHolder<>(
+ config.getMaxPerKey(),
+ config.getUnusedConnectionTimeoutMillis(),
+ key,
+ factory
+ );
+ }
+ return new LazyCreationResourceHolder<>(
+ config.getMaxPerKey(),
+ config.getUnusedConnectionTimeoutMillis(),
+ key,
+ factory
+ );
+ }
+ };
+
+ abstract <K, V> PooledResources<V> create(
+ ResourcePoolConfig config,
+ K key,
+ ResourceFactory<K, V> factory,
+ boolean eagerInitialization
+ );
+
+ @JsonValue
+ @Override
+ public String toString()
+ {
+ return StringUtils.toLowerCase(name());
+ }
+
+ @JsonCreator
+ public static Implementation fromString(String name)
+ {
+ return valueOf(StringUtils.toUpperCase(name));
+ }
+ }
+
+ /**
+ * The resources of a single key.
+ */
+ private abstract static class PooledResources<V> implements Closeable
+ {
+ final Counters counters = new Counters();
+
+ /**
+ * Takes a resource, blocking until one is free. Null if closed or
interrupted.
+ */
+ @Nullable
+ abstract V get();
+
+ /**
+ * Resources handed to a caller and not given back yet.
+ */
+ abstract int getUsedCount();
+
+ /**
+ * Resources parked for the next caller.
+ */
+ abstract int getIdleCount();
+
+ final Stats drainStats()
+ {
+ return counters.drain(getUsedCount(), getIdleCount());
+ }
+
+
+ /**
+ * Returns a resource taken from {@link #get()}, freeing the slot it
occupied. An implementation may discard the
+ * resource instead of pooling it.
+ */
+ abstract void giveBack(V object);
+
+ /**
+ * Discards every idle resource and releases every waiting taker. Lent
resources are closed on {@link #giveBack}.
+ */
+ @Override
+ public abstract void close();
+ }
+
private static class EagerCreationResourceHolder<K, V> extends
LazyCreationResourceHolder<K, V>
{
private EagerCreationResourceHolder(
@@ -183,7 +423,7 @@ public class ResourcePool<K, V> implements Closeable
new ResourceHolder<>(
System.currentTimeMillis(),
Preconditions.checkNotNull(
- factory.generate(key),
+ this.factory.generate(key),
"factory.generate(key)"
)
)
@@ -205,11 +445,11 @@ public class ResourcePool<K, V> implements Closeable
}
}
- private static class ResourceHolderPerKey<K, V> implements Closeable
+ private static class ResourceHolderPerKey<K, V> extends PooledResources<V>
{
protected final int maxSize;
+ protected final ResourceFactory<K, V> factory;
private final K key;
- private final ResourceFactory<K, V> factory;
private final long unusedResourceTimeoutMillis;
// Hold previously created / returned resources
protected final ArrayDeque<ResourceHolder<V>> resourceHolderList;
@@ -226,7 +466,7 @@ public class ResourcePool<K, V> implements Closeable
{
this.maxSize = maxSize;
this.key = key;
- this.factory = factory;
+ this.factory = new CountingResourceFactory<>(factory, counters);
this.unusedResourceTimeoutMillis = unusedResourceTimeoutMillis;
this.resourceHolderList = new ArrayDeque<>();
}
@@ -237,6 +477,7 @@ public class ResourcePool<K, V> implements Closeable
* Try to return a previously created resource if it isGood(). Else,
generate a new resource
*/
@Nullable
+ @Override
V get()
{
final V poolVal;
@@ -265,6 +506,7 @@ public class ResourcePool<K, V> implements Closeable
poolVal = holder.getResource();
if (System.currentTimeMillis() - holder.getLastAccessedTime() >
unusedResourceTimeoutMillis) {
expired = true;
+ counters.timedOut.incrementAndGet();
}
}
numLentResources++;
@@ -297,6 +539,7 @@ public class ResourcePool<K, V> implements Closeable
return retVal;
}
+ @Override
void giveBack(V object)
{
Preconditions.checkNotNull(object, "object");
@@ -338,6 +581,18 @@ public class ResourcePool<K, V> implements Closeable
}
}
+ @Override
+ synchronized int getUsedCount()
+ {
+ return numLentResources;
+ }
+
+ @Override
+ synchronized int getIdleCount()
+ {
+ return resourceHolderList.size();
+ }
+
private boolean holderListContains(V object)
{
return resourceHolderList.stream().anyMatch(a ->
a.getResource().equals(object));
@@ -355,6 +610,215 @@ public class ResourcePool<K, V> implements Closeable
}
}
+ private static class AdaptiveResourceHolderPerKey<K, V> extends
PooledResources<V>
+ {
+ private static final double BAD_RESOURCE_WARN_PERIOD_SECONDS = 5;
+
+ private final int maxSize;
+ private final K key;
+ private final ResourceFactory<K, V> factory;
+ private final long unusedResourceTimeoutMillis;
+ private final Semaphore permits;
+ private final Deque<ResourceHolder<V>> idleResources = new
ConcurrentLinkedDeque<>();
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final AtomicInteger lentResources = new AtomicInteger();
+ private final RateLimiter badResourceWarnings = RateLimiter.create(1 /
BAD_RESOURCE_WARN_PERIOD_SECONDS);
+
+ private AdaptiveResourceHolderPerKey(
+ int maxSize,
+ long unusedResourceTimeoutMillis,
+ K key,
+ ResourceFactory<K, V> factory
+ )
+ {
+ this.maxSize = maxSize;
+ this.key = key;
+ this.factory = new CountingResourceFactory<>(factory, counters);
+ this.unusedResourceTimeoutMillis = unusedResourceTimeoutMillis;
+ this.permits = new Semaphore(maxSize);
+ }
+
+ /**
+ * Fills the pool to its maximum size, closing what it created if that
fails.
+ */
+ void preload()
+ {
+ try {
+ for (int i = 0; i < maxSize; i++) {
+ idleResources.addLast(new
ResourceHolder<>(System.currentTimeMillis(), generate()));
+ }
+ }
+ catch (Throwable t) {
+ closeIdleResources();
+ throw t;
+ }
+ }
+
+ @Nullable
+ @Override
+ V get()
+ {
+ if (!acquirePermit()) {
+ return null;
+ }
+
+ boolean lent = false;
+ try {
+ V resource = takeIdleResource();
+ if (resource == null) {
+ resource = createResource();
+ }
+ lent = true;
+ lentResources.incrementAndGet();
+ return resource;
+ }
+ finally {
+ if (!lent) {
+ permits.release();
+ }
+ }
+ }
+
+ @Override
+ int getUsedCount()
+ {
+ return lentResources.get();
+ }
+
+ @Override
+ int getIdleCount()
+ {
+ return idleResources.size();
+ }
+
+ @Override
+ void giveBack(V object)
+ {
+ Preconditions.checkNotNull(object, "object");
+
+ try {
+ if (closed.get()) {
+ log.info("giveBack called after being closed. key[%s]", key);
+ closeQuietly(object);
+ return;
+ }
+ if (!isGood(object)) {
+ closeQuietly(object);
+ return;
+ }
+ idleResources.addLast(new ResourceHolder<>(System.currentTimeMillis(),
object));
+ }
+ finally {
+ lentResources.decrementAndGet();
+ permits.release();
+ }
+
+ if (closed.get()) {
+ // close() may have drained the idle resources before this one was
parked.
+ closeIdleResources();
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ if (closed.compareAndSet(false, true)) {
+ permits.release(Integer.MAX_VALUE / 2);
+ closeIdleResources();
+ }
+ }
+
+ private boolean acquirePermit()
+ {
+ try {
+ permits.acquire();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+
+ if (closed.get()) {
+ log.info("get() called even though I'm closed. key[%s]", key);
+ permits.release();
+ return false;
+ }
+ return true;
+ }
+
+ @Nullable
+ private V takeIdleResource()
+ {
+ final long expiredBefore = System.currentTimeMillis() -
unusedResourceTimeoutMillis;
+ for (ResourceHolder<V> holder = idleResources.pollFirst(); holder !=
null; holder = idleResources.pollFirst()) {
+ final V resource = holder.getResource();
+ if (holder.getLastAccessedTime() < expiredBefore) {
+ counters.timedOut.incrementAndGet();
+ closeQuietly(resource);
+ } else if (isGood(resource)) {
+ return resource;
+ } else {
+ closeQuietly(resource);
+ }
+ }
+ return null;
+ }
+
+ private V createResource()
+ {
+ final V resource = generate();
+ if (isGood(resource)) {
+ return resource;
+ }
+ if (badResourceWarnings.tryAcquire()) {
+ log.warn(
+ "Handing over resource[%s] at key[%s] that failed its health check
right after being created, it may be"
+ + " bad. Further such warnings about this key are suppressed for
[%s] seconds.",
+ resource,
+ key,
+ BAD_RESOURCE_WARN_PERIOD_SECONDS
+ );
+ }
+ return resource;
+ }
+
+ /**
+ * The verdict of {@link ResourceFactory#isGood}, closing the resource if
the check itself throws.
+ */
+ private boolean isGood(V resource)
+ {
+ try {
+ return factory.isGood(resource);
+ }
+ catch (Throwable t) {
+ closeQuietly(resource);
+ throw t;
+ }
+ }
+
+ private V generate()
+ {
+ return Preconditions.checkNotNull(factory.generate(key),
"factory.generate(key)");
+ }
+
+ private void closeIdleResources()
+ {
+ for (ResourceHolder<V> holder = idleResources.pollFirst(); holder !=
null; holder = idleResources.pollFirst()) {
+ closeQuietly(holder.getResource());
+ }
+ }
+
+ private void closeQuietly(V resource)
+ {
+ try {
+ factory.close(resource);
+ }
+ catch (Exception e) {
+ log.warn(e, "Failed to close resource at key[%s]", key);
+ }
+ }
+ }
+
private static class ResourceHolder<V>
{
private final long lastAccessedTime;
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePoolConfig.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePoolConfig.java
index 81ce018bd6f..f254bb2f759 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePoolConfig.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/pool/ResourcePoolConfig.java
@@ -25,14 +25,25 @@ public class ResourcePoolConfig
{
private final int maxPerKey;
private final long unusedConnectionTimeoutMillis;
+ private final ResourcePool.Implementation poolImplementation;
public ResourcePoolConfig(
int maxPerKey,
long unusedConnectionTimeoutMillis
)
+ {
+ this(maxPerKey, unusedConnectionTimeoutMillis,
ResourcePool.Implementation.ADAPTIVE);
+ }
+
+ public ResourcePoolConfig(
+ int maxPerKey,
+ long unusedConnectionTimeoutMillis,
+ ResourcePool.Implementation poolImplementation
+ )
{
this.maxPerKey = maxPerKey;
this.unusedConnectionTimeoutMillis = unusedConnectionTimeoutMillis;
+ this.poolImplementation = poolImplementation;
}
@Deprecated
@@ -61,4 +72,9 @@ public class ResourcePoolConfig
{
return unusedConnectionTimeoutMillis;
}
+
+ public ResourcePool.Implementation getPoolImplementation()
+ {
+ return poolImplementation;
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/java/util/http/client/pool/ResourcePoolTest.java
b/processing/src/test/java/org/apache/druid/java/util/http/client/pool/ResourcePoolTest.java
index 7b262924ae3..f991e11370b 100644
---
a/processing/src/test/java/org/apache/druid/java/util/http/client/pool/ResourcePoolTest.java
+++
b/processing/src/test/java/org/apache/druid/java/util/http/client/pool/ResourcePoolTest.java
@@ -19,493 +19,694 @@
package org.apache.druid.java.util.http.client.pool;
+import com.google.common.util.concurrent.UncheckedExecutionException;
import org.apache.druid.java.util.common.ISE;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.testing.junit.LoggerCaptureExtension;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
/**
+ * The contract every {@link ResourcePool.Implementation} owes its callers,
followed by the promises each
+ * implementation makes on its own.
*/
public class ResourcePoolTest
{
- ResourceFactory<String, String> resourceFactory;
- ResourcePool<String, String> pool;
+ private static final long NEVER_EXPIRES = TimeUnit.MINUTES.toMillis(5);
+ private static final long EXPIRES_QUICKLY = 100;
+
+ @RegisterExtension
+ public final LoggerCaptureExtension logger = new
LoggerCaptureExtension(ResourcePool.class);
+
+ private TestResourceFactory factory;
@BeforeEach
public void setUp()
{
- setUpPool(true);
+ factory = new TestResourceFactory();
}
- public void setUpPoolWithoutEagerInitialization()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void testReturnedResourceIsReused(ResourcePool.Implementation
implementation)
{
- setUpPool(false);
- }
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, false);
- public void setUpPool(boolean eagerInitialization)
- {
- resourceFactory = (ResourceFactory<String, String>)
EasyMock.createMock(ResourceFactory.class);
+ final ResourceContainer<String> first = pool.take("billy");
+ Assertions.assertEquals("billy#0", first.get());
+ first.returnResource();
- EasyMock.replay(resourceFactory);
- pool = new ResourcePool<>(
- resourceFactory,
- new ResourcePoolConfig(2, TimeUnit.MINUTES.toMillis(4)),
- eagerInitialization
- );
+ final ResourceContainer<String> second = pool.take("billy");
+ Assertions.assertEquals("billy#0", second.get(), "a returned resource is
handed out again");
+ second.returnResource();
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
+ Assertions.assertEquals(List.of("billy#0"), factory.opened(), "no second
connection is opened");
+ Assertions.assertEquals(List.of(), factory.closed());
}
- @Test
- public void testSanity()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testEagerInitializationFillsAKeyOnItsFirstTake(ResourcePool.Implementation
implementation)
{
- primePool();
- EasyMock.replay(resourceFactory);
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, true);
+ Assertions.assertEquals(List.of(), factory.opened(), "an untouched key
costs nothing");
+
+ final ResourceContainer<String> billy = pool.take("billy");
+ Assertions.assertEquals(List.of("billy#0", "billy#1"), factory.opened());
+ Assertions.assertEquals("billy#0", billy.get());
+ billy.returnResource();
}
- @Test
- public void testTakeOnce_lazy()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testTakeBlocksWhileEveryResourceIsLent(ResourcePool.Implementation
implementation) throws Exception
{
- setUpPoolWithoutEagerInitialization();
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, false);
+
+ final ResourceContainer<String> first = pool.take("billy");
+ final ResourceContainer<String> second = pool.take("billy");
- EasyMock.expect(resourceFactory.generate("billy")).andAnswer(new
StringIncrementingAnswer("billy")).times(1);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
+ final BackgroundTake third = BackgroundTake.start(pool, "billy");
+ Assertions.assertTrue(third.isBlocked(), "a third take while both
resources are lent");
- ResourceContainer<String> billyString = pool.take("billy");
- Assertions.assertEquals("billy0", billyString.get());
+ first.returnResource();
+ Assertions.assertEquals("billy#0", third.awaitResource(), "the returned
resource unblocks the waiting take");
+ third.release();
+ second.returnResource();
- billyString.returnResource();
+ Assertions.assertEquals(2, factory.opened().size(), "the pool never opens
more than maxPerKey");
}
- @Test
- public void testTakeAfterReturn_lazy()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testUnhealthyResourceIsClosedAndReplaced(ResourcePool.Implementation
implementation)
{
- setUpPoolWithoutEagerInitialization();
-
- // Generate and check before return
- EasyMock.expect(resourceFactory.generate("billy")).andAnswer(new
StringIncrementingAnswer("billy")).times(1);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- // Only check since there's no need to generate after return
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
-
- ResourceContainer<String> billyString = pool.take("billy");
- Assertions.assertEquals("billy0", billyString.get());
-
- billyString.returnResource();
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, false);
+ pool.take("billy").returnResource();
- billyString = pool.take("billy");
- Assertions.assertEquals("billy0", billyString.get());
+ factory.markUnhealthy("billy#0");
- billyString.returnResource();
+ final ResourceContainer<String> replacement = pool.take("billy");
+ Assertions.assertEquals("billy#1", replacement.get());
+ Assertions.assertEquals(List.of("billy#0"), factory.closed());
+ replacement.returnResource();
}
- @Test
- public void testTakeAfterFailure()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testExpiredResourceIsDiscardedWithoutTouchingOtherKeys(ResourcePool.Implementation
implementation)
+ throws Exception
{
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy0");
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy1");
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
EXPIRES_QUICKLY, false);
+ pool.take("billy").returnResource();
+ awaitExpiry();
+ pool.take("sally").returnResource();
- EasyMock.expect(resourceFactory.isGood("billy0")).andThrow(new
RuntimeException("blah"));
+ final ResourceContainer<String> billy = pool.take("billy");
+ Assertions.assertEquals("billy#1", billy.get(), "the expired resource must
not be handed out");
+ billy.returnResource();
- EasyMock.expect(resourceFactory.isGood("billy1")).andThrow(new
RuntimeException("blah"));
+ final ResourceContainer<String> sally = pool.take("sally");
+ Assertions.assertEquals("sally#0", sally.get(), "expiry of one key must
not touch another key");
+ sally.returnResource();
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy2");
- EasyMock.expect(resourceFactory.isGood("billy2")).andReturn(true);
+ Assertions.assertEquals(List.of("billy#0"), factory.closed());
+ }
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy3");
- EasyMock.expect(resourceFactory.isGood("billy3")).andReturn(true);
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testCloseDiscardsIdleResourcesAndRefusesFurtherTakes(ResourcePool.Implementation
implementation)
+ {
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, false);
+ pool.take("billy").returnResource();
- EasyMock.expect(resourceFactory.isGood("billy2")).andReturn(true);
+ pool.close();
- EasyMock.expect(resourceFactory.isGood("billy3")).andReturn(true);
+ Assertions.assertEquals(List.of("billy#0"), factory.closed());
+ Assertions.assertNull(pool.take("billy"), "take after close");
+ }
- EasyMock.expect(resourceFactory.isGood("billy2")).andReturn(true);
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void testCloseUnblocksAWaitingTake(ResourcePool.Implementation
implementation) throws Exception
+ {
+ final ResourcePool<String, String> pool = createPool(implementation, 1,
NEVER_EXPIRES, false);
+ final ResourceContainer<String> lent = pool.take("billy");
- EasyMock.replay(resourceFactory);
- // numLentResources == 0, resourceHolderList.size() == 2
+ final BackgroundTake waiter = BackgroundTake.start(pool, "billy");
+ Assertions.assertTrue(waiter.isBlocked(), "a take while the only resource
is lent");
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources == 0, resourceHolderList.size() == 1
+ pool.close();
+ Assertions.assertNull(waiter.awaitResource(), "a take unblocked by close()
hands out nothing");
+ waiter.release();
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources == 0, resourceHolderList.size() == 0
+ lent.returnResource();
+ Assertions.assertEquals(List.of("billy#0"), factory.closed(), "a resource
returned after close is closed");
+ }
+
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void testReturningTwiceRepaysOneSlotOnly(ResourcePool.Implementation
implementation) throws Exception
+ {
+ final ResourcePool<String, String> pool = createPool(implementation, 1,
NEVER_EXPIRES, false);
+ final ResourceContainer<String> billy = pool.take("billy");
+ billy.returnResource();
+ billy.returnResource();
- ResourceContainer<String> a = pool.take("billy");
- // numLentResources == 1, resourceHolderList.size() == 0
+ Assertions.assertThrows(IllegalStateException.class, billy::get, "a
returned container cannot be read again");
- ResourceContainer<String> b = pool.take("billy");
- // numLentResources == 2, resourceHolderList.size() == 0
+ final ResourceContainer<String> reused = pool.take("billy");
+ final BackgroundTake extra = BackgroundTake.start(pool, "billy");
+ Assertions.assertTrue(extra.isBlocked(), "the second return must not have
created a second slot");
- a.returnResource();
- // numLentResources = 1, resourceHolderList.size() == 1
+ reused.returnResource();
+ Assertions.assertEquals("billy#0", extra.awaitResource());
+ extra.release();
+ }
- a = pool.take("billy");
- // numLentResources = 2, resourceHolderList.size() == 0
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testFailedGenerateDoesNotConsumeASlot(ResourcePool.Implementation
implementation) throws Exception
+ {
+ final ResourcePool<String, String> pool = createPool(implementation, 1,
NEVER_EXPIRES, false);
+ factory.failGenerate(new ISE("no more billies"));
- b.returnResource();
- // numLentResources = 1, resourceHolderList.size() == 1
+ Assertions.assertThrows(ISE.class, () -> pool.take("billy"));
- a.returnResource();
- // numLentResources = 0, resourceHolderList.size() == 2
+ factory.healGenerate();
+ final BackgroundTake next = BackgroundTake.start(pool, "billy");
+ Assertions.assertEquals("billy#0", next.awaitResource(), "the failed take
must give its slot back");
+ next.release();
}
- @Test
- public void testTakeAfterFailure_lazy()
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testFailedHealthCheckDoesNotConsumeASlot(ResourcePool.Implementation
implementation) throws Exception
{
- setUpPoolWithoutEagerInitialization();
+ final ResourcePool<String, String> pool = createPool(implementation, 1,
NEVER_EXPIRES, false);
+ pool.take("billy").returnResource();
+ factory.failHealthCheck("billy#0", new ISE("health check blew up"));
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy0");
- EasyMock.expect(resourceFactory.isGood("billy0")).andThrow(new
RuntimeException("blah"));
+ Assertions.assertThrows(ISE.class, () -> pool.take("billy"));
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy1");
- EasyMock.expect(resourceFactory.isGood("billy1")).andThrow(new
RuntimeException("blah"));
-
- EasyMock.expect(resourceFactory.generate("billy")).andThrow(new
RuntimeException("blah"));
+ final BackgroundTake next = BackgroundTake.start(pool, "billy");
+ Assertions.assertEquals("billy#1", next.awaitResource(), "the failed take
must give its slot back");
+ next.release();
+ }
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy3");
- EasyMock.expect(resourceFactory.isGood("billy3")).andReturn(true);
+ /**
+ * Under contention, with expiry constantly evicting resources, the pool
never lends one resource to two callers,
+ * never exceeds maxPerKey, never hands out a closed resource and never
loses one.
+ */
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testConcurrentTakersNeverShareAResource(ResourcePool.Implementation
implementation) throws Exception
+ {
+ final int maxPerKey = 4;
+ final int takers = 8;
+ final ResourcePool<String, String> pool = createPool(implementation,
maxPerKey, 5, false);
+
+ final Set<String> lent = ConcurrentHashMap.newKeySet();
+ final AtomicInteger lentCount = new AtomicInteger();
+ final AtomicReference<String> misuse = new AtomicReference<>();
+
+ final ExecutorService exec = Execs.multiThreaded(takers,
"resource-pool-stress-%d");
+ final List<Future<?>> futures = new ArrayList<>();
+ for (int i = 0; i < takers; i++) {
+ futures.add(exec.submit(() -> {
+ for (int j = 0; j < 200; j++) {
+ final ResourceContainer<String> container = pool.take("billy");
+ final String resource = container.get();
+ if (!lent.add(resource)) {
+ misuse.compareAndSet(null, "lent to two takers at once: " +
resource);
+ }
+ if (lentCount.incrementAndGet() > maxPerKey) {
+ misuse.compareAndSet(null, "more than " + maxPerKey + " lent at
once");
+ }
+ if (factory.isClosed(resource)) {
+ misuse.compareAndSet(null, "lent after being closed: " + resource);
+ }
+ lentCount.decrementAndGet();
+ lent.remove(resource);
+ container.returnResource();
+ }
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ exec.shutdown();
+ pool.close();
- EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy4");
- EasyMock.expect(resourceFactory.isGood("billy4")).andReturn(true);
+ Assertions.assertNull(misuse.get());
+ Assertions.assertEquals(Set.copyOf(factory.opened()),
Set.copyOf(factory.closed()), "every resource is closed");
+ Assertions.assertEquals(factory.opened().size(), factory.closed().size(),
"no resource is closed twice");
+ }
- EasyMock.expect(resourceFactory.isGood("billy3")).andReturn(true);
+ /**
+ * One take() discards every expired resource it walks past and opens a
single replacement, so a pool that ran hot
+ * shrinks back to what the traffic needs instead of re-handshaking up to
maxPerKey one take at a time.
+ */
+ @Test
+ public void testAdaptivePurgesEveryExpiredResourceInOneTake() throws
Exception
+ {
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 3, EXPIRES_QUICKLY, true);
+ pool.take("billy").returnResource();
+ awaitExpiry();
- EasyMock.expect(resourceFactory.isGood("billy4")).andReturn(true);
+ final ResourceContainer<String> fresh = pool.take("billy");
+ Assertions.assertEquals("billy#3", fresh.get());
+ fresh.returnResource();
- EasyMock.expect(resourceFactory.isGood("billy3")).andThrow(new
RuntimeException("blah"));
+ Assertions.assertEquals(Set.of("billy#0", "billy#1", "billy#2"),
Set.copyOf(factory.closed()));
+ Assertions.assertEquals(4, factory.opened().size(), "exactly one
replacement is opened");
+ }
- EasyMock.replay(resourceFactory);
- // numLentResources == 0, resourceHolderList.size() == 0
+ /**
+ * A peer restart kills every pooled resource at once, long before any of
them expires. One take() must clear them
+ * all rather than discovering them one take - and one handshake on a
caller's thread - at a time.
+ */
+ @Test
+ public void testAdaptivePurgesEveryDeadResourceInOneTake()
+ {
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 3, NEVER_EXPIRES, true);
+ pool.take("billy").returnResource();
+ factory.markUnhealthy("billy#0", "billy#1", "billy#2");
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources == 0, resourceHolderList.size() == 0
+ final ResourceContainer<String> fresh = pool.take("billy");
+ Assertions.assertEquals("billy#3", fresh.get());
+ fresh.returnResource();
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources == 0, resourceHolderList.size() == 0
+ Assertions.assertEquals(Set.of("billy#0", "billy#1", "billy#2"),
Set.copyOf(factory.closed()));
+ }
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources == 0, resourceHolderList.size() == 0
+ /**
+ * A resource that broke while it was lent out is closed on its way back
instead of being parked for the next
+ * taker, whose health check would only discover it a moment later.
+ */
+ @Test
+ public void testAdaptiveDiscardsABrokenResourceOnGiveBack()
+ {
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 1, NEVER_EXPIRES, false);
+ final ResourceContainer<String> lent = pool.take("billy");
+ factory.markUnhealthy("billy#0");
- ResourceContainer<String> a = pool.take("billy");
- // numLentResources == 1, resourceHolderList.size() == 0
+ lent.returnResource();
+ Assertions.assertEquals(List.of("billy#0"), factory.closed());
- ResourceContainer<String> b = pool.take("billy");
- // numLentResources == 2, resourceHolderList.size() == 0
+ final ResourceContainer<String> fresh = pool.take("billy");
+ Assertions.assertEquals("billy#1", fresh.get(), "the discarded resource
repaid its slot");
+ fresh.returnResource();
+ }
- a.returnResource();
- // numLentResources == 1, resourceHolderList.size() == 1
+ /**
+ * A freshly opened resource that fails its health check is handed over
rather than replaced: opening another
+ * connection cannot mend a peer that is refusing health checks, so the
caller gets its resource and a warning is
+ * logged.
+ */
+ @Test
+ public void testAdaptiveHandsOverAFreshResourceThatFailsItsHealthCheck()
+ {
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 2, NEVER_EXPIRES, false);
+ factory.markUnhealthy("billy#0");
+
+ final ResourceContainer<String> lent = pool.take("billy");
+ Assertions.assertEquals("billy#0", lent.get(), "no replacement is opened");
+ Assertions.assertEquals(List.of("billy#0"), factory.opened(), "and none is
attempted either");
+ Assertions.assertEquals(List.of(), factory.closed(), "the handed over
resource stays open");
+ Assertions.assertTrue(
+ logger.getLogEvents().stream().anyMatch(e ->
e.getMessage().getFormattedMessage().contains("billy#0")),
+ "the caller is warned about it"
+ );
- try {
- pool.take("billy");
- }
- catch (Exception e) {
- }
- // numLentResources = 1, resourceHolderList.size() == 0
+ lent.returnResource();
+ Assertions.assertTrue(factory.isClosed("billy#0"), "and it is discarded
once it comes back");
+ }
- b.returnResource();
- // numLentResources = 0, resourceHolderList.size() == 1
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testStatsRecordWhatHappenedToTheResourcesOfAKey(ResourcePool.Implementation
implementation)
+ throws Exception
+ {
+ final ResourcePool<String, String> pool = createPool(implementation, 1,
EXPIRES_QUICKLY, false);
+ pool.take("billy").returnResource();
+ awaitExpiry();
+ pool.take("billy").returnResource();
+ pool.take("sally").returnResource();
+
+ final Map<String, ResourcePool.Stats> stats = pool.drainStats();
+ Assertions.assertEquals(Set.of("billy", "sally"), stats.keySet(), "one
entry per key");
+
+ final ResourcePool.Stats billy = stats.get("billy");
+ Assertions.assertEquals(2, billy.opened(), "opened");
+ Assertions.assertEquals(1, billy.closed(), "closed");
+ Assertions.assertEquals(1, billy.timedOut(), "timed out");
+ Assertions.assertEquals(0, billy.errored(), "errored");
+ Assertions.assertEquals(2, billy.taken(), "taken");
+ Assertions.assertEquals(2, billy.returned(), "returned");
+ Assertions.assertEquals(0, billy.used(), "nothing is lent out");
+ Assertions.assertEquals(1, billy.idle(), "the returned resource waits for
the next caller");
+
+ Assertions.assertEquals(1, stats.get("sally").opened(), "keys are counted
apart");
}
- private void primePool()
+ /**
+ * The counted events are drained as they are read, so an idle key reports
zeroes rather than the totals it
+ * reported before, while what it holds keeps being reported.
+ */
+ @ParameterizedTest
+ @EnumSource(ResourcePool.Implementation.class)
+ public void
testDrainedStatsCoverOnlyWhatHappenedSinceThePreviousDrain(ResourcePool.Implementation
implementation)
{
- EasyMock.expect(resourceFactory.generate("billy")).andAnswer(new
StringIncrementingAnswer("billy")).times(2);
- EasyMock.expect(resourceFactory.generate("sally")).andAnswer(new
StringIncrementingAnswer("sally")).times(2);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.expect(resourceFactory.isGood("sally0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
-
- ResourceContainer<String> billyString = pool.take("billy");
- ResourceContainer<String> sallyString = pool.take("sally");
- Assertions.assertEquals("billy0", billyString.get());
- Assertions.assertEquals("sally0", sallyString.get());
-
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
-
- billyString.returnResource();
- sallyString.returnResource();
+ final ResourcePool<String, String> pool = createPool(implementation, 2,
NEVER_EXPIRES, false);
+ final ResourceContainer<String> lent = pool.take("billy");
+ pool.drainStats();
+
+ final ResourcePool.Stats idle = pool.drainStats().get("billy");
+ Assertions.assertEquals(0, idle.opened(), "opened");
+ Assertions.assertEquals(0, idle.taken(), "taken");
+ Assertions.assertEquals(1, idle.used(), "the lent resource is still out");
+ Assertions.assertEquals(0, idle.idle(), "and is not parked");
+
+ lent.returnResource();
+ final ResourcePool.Stats afterReturn = pool.drainStats().get("billy");
+ Assertions.assertEquals(1, afterReturn.returned(), "returned");
+ Assertions.assertEquals(0, afterReturn.used(), "used");
+ Assertions.assertEquals(1, afterReturn.idle(), "idle");
}
@Test
- public void testFailedResource()
+ public void testStatsRecordAFailingHealthCheckAsAnError()
{
- primePool();
-
-
EasyMock.expect(resourceFactory.isGood("billy1")).andReturn(false).times(1);
- resourceFactory.close("billy1");
- EasyMock.expectLastCall();
-
EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy2").times(1);
- EasyMock.replay(resourceFactory);
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 2, NEVER_EXPIRES, false);
+ pool.take("billy").returnResource();
+ factory.failHealthCheck("billy#0", new ISE("health check blew up"));
- ResourceContainer<String> billy = pool.take("billy");
- Assertions.assertEquals("billy2", billy.get());
- billy.returnResource();
+ Assertions.assertThrows(ISE.class, () -> pool.take("billy"));
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
+ final ResourcePool.Stats stats = pool.drainStats().get("billy");
+ Assertions.assertEquals(1, stats.errored(), "errored");
+ Assertions.assertEquals(1, stats.closed(), "the resource whose check threw
is closed");
}
+ /**
+ * A resource taken off the queue and handed to nobody is closed rather than
leaked.
+ */
@Test
- public void testFaultyFailedResourceReplacement()
+ public void testAdaptiveClosesTheResourceWhoseHealthCheckThrows()
{
- primePool();
-
-
EasyMock.expect(resourceFactory.isGood("billy1")).andReturn(false).times(1);
- resourceFactory.close("billy1");
- EasyMock.expectLastCall();
- EasyMock.expect(resourceFactory.generate("billy")).andThrow(new
ISE("where's billy?")).times(1);
-
EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(false).times(1);
- resourceFactory.close("billy0");
- EasyMock.expectLastCall();
- EasyMock.expect(resourceFactory.generate("billy")).andThrow(new
ISE("where's billy?")).times(1);
-
EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy2").times(1);
- EasyMock.expect(resourceFactory.isGood("billy2")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
-
- IllegalStateException e1 = null;
- try {
- pool.take("billy");
- }
- catch (IllegalStateException e) {
- e1 = e;
- }
- Assertions.assertNotNull(e1, "exception");
- Assertions.assertEquals("where's billy?", e1.getMessage());
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 2, NEVER_EXPIRES, false);
+ pool.take("billy").returnResource();
+ factory.failHealthCheck("billy#0", new ISE("health check blew up"));
- IllegalStateException e2 = null;
- try {
- pool.take("billy");
- }
- catch (IllegalStateException e) {
- e2 = e;
- }
- Assertions.assertNotNull(e2, "exception");
- Assertions.assertEquals("where's billy?", e2.getMessage());
+ Assertions.assertThrows(ISE.class, () -> pool.take("billy"));
+ Assertions.assertEquals(List.of("billy#0"), factory.closed());
+ }
- ResourceContainer<String> billy = pool.take("billy");
- Assertions.assertEquals("billy2", billy.get());
- billy.returnResource();
+ /**
+ * A resource that fails to close while being evicted does not fail the take
that evicted it, and does not strand
+ * the resources queued behind it.
+ */
+ @Test
+ public void testAdaptiveSurvivesAFailingClose() throws Exception
+ {
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 2, EXPIRES_QUICKLY, true);
+ pool.take("billy").returnResource();
+ awaitExpiry();
+ // billy#1 is at the head of the queue, so it is the first one evicted.
+ factory.failClose("billy#1", new ISE("close blew up"));
+
+ final ResourceContainer<String> fresh = pool.take("billy");
+ Assertions.assertEquals("billy#2", fresh.get());
+ fresh.returnResource();
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
+ Assertions.assertEquals(Set.of("billy#0", "billy#1"),
Set.copyOf(factory.closed()));
}
+ /**
+ * Resources opened before a failing eager initialization are closed - the
half-built holder never reaches the
+ * cache, so nothing can ever reach them again.
+ */
@Test
- public void testTakeMoreThanAllowed() throws Exception
+ public void
testAdaptiveClosesResourcesOpenedBeforeAFailedEagerInitialization()
{
- primePool();
- EasyMock.expect(resourceFactory.isGood("billy1")).andReturn(true).times(1);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
-
- CountDownLatch latch1 = new CountDownLatch(1);
- CountDownLatch latch2 = new CountDownLatch(1);
- CountDownLatch latch3 = new CountDownLatch(1);
-
- MyThread billy1Thread = new MyThread(latch1, "billy");
- billy1Thread.start();
- billy1Thread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
- MyThread billy0Thread = new MyThread(latch2, "billy");
- billy0Thread.start();
- billy0Thread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
-
- MyThread blockedThread = new MyThread(latch3, "billy");
- blockedThread.start();
-
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
-
- latch2.countDown();
- blockedThread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
-
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
-
- latch1.countDown();
- latch3.countDown();
-
- Assertions.assertEquals("billy1", billy1Thread.getValue());
- Assertions.assertEquals("billy0", billy0Thread.getValue());
- Assertions.assertEquals("billy0", blockedThread.getValue());
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 3, NEVER_EXPIRES, true);
+ factory.failGenerateAfter(2, new ISE("no more billies"));
+
+ final Exception thrown =
Assertions.assertThrows(UncheckedExecutionException.class, () ->
pool.take("billy"));
+ Assertions.assertEquals("no more billies", thrown.getCause().getMessage());
+ Assertions.assertEquals(List.of("billy#0", "billy#1"), factory.closed());
}
+ /**
+ * A null from {@link ResourceFactory#generate} fails the take instead of
producing a container that blows up on
+ * {@link ResourceContainer#returnResource} and never repays its lent slot.
+ */
@Test
- public void testCloseUnblocks() throws InterruptedException
+ public void testAdaptiveRejectsANullResource()
{
- primePool();
- EasyMock.expect(resourceFactory.isGood("billy1")).andReturn(true).times(1);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- resourceFactory.close("sally1");
- EasyMock.expectLastCall().times(1);
- resourceFactory.close("sally0");
- EasyMock.expectLastCall().times(1);
- EasyMock.replay(resourceFactory);
- CountDownLatch latch1 = new CountDownLatch(1);
- CountDownLatch latch2 = new CountDownLatch(1);
- CountDownLatch latch3 = new CountDownLatch(1);
-
- MyThread billy1Thread = new MyThread(latch1, "billy");
- billy1Thread.start();
- billy1Thread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
- MyThread billy0Thread = new MyThread(latch2, "billy");
- billy0Thread.start();
- billy0Thread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
-
- MyThread blockedThread = new MyThread(latch3, "billy");
- blockedThread.start();
- blockedThread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
- pool.close();
+ final ResourcePool<String, String> pool =
createPool(ResourcePool.Implementation.ADAPTIVE, 2, NEVER_EXPIRES, false);
+ factory.generateNull();
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
- // billy0Thread calling ResourceContainer.returnResource() will result
- // in a call to resourceFactory.close() when latch2 is triggered
- resourceFactory.close("billy0");
- EasyMock.expectLastCall().once();
- EasyMock.replay(resourceFactory);
-
- latch2.countDown();
- blockedThread.waitForValueToBeGotten(1, TimeUnit.SECONDS);
- // wait for billy0Thread to have called resourceFactory.close() to avoid
race
- // between billy0Thread calling it and verify() checking for the call
- billy0Thread.join();
-
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
-
- latch1.countDown();
- latch3.countDown();
-
- Assertions.assertEquals("billy1", billy1Thread.getValue());
- Assertions.assertEquals("billy0", billy0Thread.getValue());
- blockedThread.join();
- // pool returns null after close
- Assertions.assertEquals(null, blockedThread.getValue());
+ Assertions.assertThrows(NullPointerException.class, () ->
pool.take("billy"));
}
+ /**
+ * The retaining pool keeps its high-water mark: an expired resource is
replaced one for one, and the resources
+ * queued behind it stay parked until their own turn comes.
+ */
@Test
- public void testTimedOutResource() throws Exception
+ public void testRetainingReplacesAnExpiredResourceOneForOne() throws
Exception
{
- resourceFactory = (ResourceFactory<String, String>)
EasyMock.createMock(ResourceFactory.class);
+ final ResourcePool<String, String> pool =
+ createPool(ResourcePool.Implementation.RETAINING, 2, EXPIRES_QUICKLY,
true);
+ pool.take("billy").returnResource();
+ awaitExpiry();
+
+ final ResourceContainer<String> fresh = pool.take("billy");
+ Assertions.assertEquals("billy#2", fresh.get());
+ fresh.returnResource();
+
+ Assertions.assertEquals(List.of("billy#1"), factory.closed(), "only the
resource at the head is discarded");
+ Assertions.assertEquals(3, factory.opened().size(), "the expired resource
is replaced one for one");
+ }
- pool = new ResourcePool<>(
- resourceFactory,
- new ResourcePoolConfig(2, TimeUnit.MILLISECONDS.toMillis(10)),
- true
+ private ResourcePool<String, String> createPool(
+ ResourcePool.Implementation implementation,
+ int maxPerKey,
+ long unusedResourceTimeoutMillis,
+ boolean eagerInitialization
+ )
+ {
+ return new ResourcePool<>(
+ factory,
+ new ResourcePoolConfig(maxPerKey, unusedResourceTimeoutMillis,
implementation),
+ eagerInitialization
);
+ }
- EasyMock.expect(resourceFactory.generate("billy")).andAnswer(new
StringIncrementingAnswer("billy")).times(2);
- EasyMock.expect(resourceFactory.isGood("billy0")).andReturn(true).times(1);
- EasyMock.replay(resourceFactory);
+ /**
+ * Waits until every idle resource of an {@link #EXPIRES_QUICKLY} pool has
gone stale.
+ */
+ private static void awaitExpiry() throws InterruptedException
+ {
+ Thread.sleep(EXPIRES_QUICKLY * 3);
+ }
- ResourceContainer<String> billyString = pool.take("billy");
- Assertions.assertEquals("billy0", billyString.get());
+ /**
+ * A take() running on its own thread, so a test can watch it block and
decide when it gives its resource back.
+ */
+ private static class BackgroundTake
+ {
+ private final CountDownLatch taken = new CountDownLatch(1);
+ private final CountDownLatch released = new CountDownLatch(1);
+ private final Thread thread;
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
+ private volatile ResourceContainer<String> container;
+ private volatile Throwable failure;
- billyString.returnResource();
+ static BackgroundTake start(ResourcePool<String, String> pool, String key)
+ {
+ return new BackgroundTake(pool, key);
+ }
- //make sure resources have been timed out.
- Thread.sleep(100);
+ private BackgroundTake(ResourcePool<String, String> pool, String key)
+ {
+ this.thread = new Thread(
+ () -> {
+ try {
+ container = pool.take(key);
+ taken.countDown();
+ released.await();
+ if (container != null) {
+ container.returnResource();
+ }
+ }
+ catch (Throwable t) {
+ failure = t;
+ }
+ finally {
+ taken.countDown();
+ }
+ },
+ "background-take-" + key
+ );
+ thread.setDaemon(true);
+ thread.start();
+ }
-
EasyMock.expect(resourceFactory.generate("billy")).andReturn("billy1").times(1);
- resourceFactory.close("billy1");
- EasyMock.replay(resourceFactory);
+ boolean isBlocked() throws InterruptedException
+ {
+ return !taken.await(200, TimeUnit.MILLISECONDS);
+ }
- ResourceContainer<String> billy = pool.take("billy");
- Assertions.assertEquals("billy1", billy.get());
- billy.returnResource();
+ /**
+ * The resource taken, or null if the pool handed out nothing. Fails if
the take never completes - which is what
+ * a leaked slot looks like, since the pool blocks once every slot is
handed out.
+ */
+ @Nullable
+ String awaitResource() throws InterruptedException
+ {
+ Assertions.assertTrue(taken.await(5, TimeUnit.SECONDS), "the take
completed");
+ return container == null ? null : container.get();
+ }
- EasyMock.verify(resourceFactory);
- EasyMock.reset(resourceFactory);
+ /**
+ * Gives the resource back and asserts that neither taking nor returning
threw.
+ */
+ void release() throws InterruptedException
+ {
+ released.countDown();
+ thread.join(TimeUnit.SECONDS.toMillis(5));
+ Assertions.assertNull(failure, "the background take failed");
+ }
}
- private static class StringIncrementingAnswer implements IAnswer<String>
+ /**
+ * Hands out resources named {@code key#n} and records what the pool did
with them. Faults are injected per
+ * resource, so a test names the resource it wants broken rather than
scripting a sequence of calls.
+ */
+ private static class TestResourceFactory implements ResourceFactory<String,
String>
{
- int count = 0;
- private String string;
+ private final ConcurrentMap<String, AtomicInteger> sequences = new
ConcurrentHashMap<>();
+ private final List<String> opened = Collections.synchronizedList(new
ArrayList<>());
+ private final List<String> closed = Collections.synchronizedList(new
ArrayList<>());
+ private final Set<String> unhealthy = ConcurrentHashMap.newKeySet();
+ private final ConcurrentMap<String, RuntimeException> healthCheckFaults =
new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, RuntimeException> closeFaults = new
ConcurrentHashMap<>();
+ private final AtomicInteger generatesBeforeFault = new AtomicInteger();
- public StringIncrementingAnswer(String string)
+ private volatile RuntimeException generateFault;
+ private volatile boolean generateNull;
+
+ @Override
+ public String generate(String key)
{
- this.string = string;
+ if (generateFault != null && generatesBeforeFault.getAndDecrement() <=
0) {
+ throw generateFault;
+ }
+ if (generateNull) {
+ return null;
+ }
+ final String resource = key + "#" + sequences.computeIfAbsent(key, k ->
new AtomicInteger()).getAndIncrement();
+ opened.add(resource);
+ return resource;
}
@Override
- public String answer()
+ public boolean isGood(String resource)
{
- return string + count++;
+ final RuntimeException fault = healthCheckFaults.get(resource);
+ if (fault != null) {
+ throw fault;
+ }
+ return !unhealthy.contains(resource);
}
- }
- private class MyThread extends Thread
- {
- private final CountDownLatch gotValueLatch = new CountDownLatch(1);
+ @Override
+ public void close(String resource)
+ {
+ closed.add(resource);
+ final RuntimeException fault = closeFaults.get(resource);
+ if (fault != null) {
+ throw fault;
+ }
+ }
- private final CountDownLatch latch1;
- private String resourceName;
+ /**
+ * Every resource ever opened, in the order they were opened.
+ */
+ List<String> opened()
+ {
+ return List.copyOf(opened);
+ }
- volatile String value = null;
+ /**
+ * Every close, in the order they happened - a resource closed twice
appears twice.
+ */
+ List<String> closed()
+ {
+ return List.copyOf(closed);
+ }
- public MyThread(CountDownLatch latch1, String resourceName)
+ boolean isClosed(String resource)
{
- this.latch1 = latch1;
- this.resourceName = resourceName;
+ return closed.contains(resource);
}
- @Override
- public void run()
+ void markUnhealthy(String... resources)
{
- ResourceContainer<String> resourceContainer = pool.take(resourceName);
- value = resourceContainer.get();
- gotValueLatch.countDown();
- try {
- latch1.await();
- }
- catch (InterruptedException e) {
+ unhealthy.addAll(List.of(resources));
+ }
- }
- resourceContainer.returnResource();
+ void failHealthCheck(String resource, RuntimeException fault)
+ {
+ healthCheckFaults.put(resource, fault);
+ }
+
+ void failClose(String resource, RuntimeException fault)
+ {
+ closeFaults.put(resource, fault);
+ }
+
+ void failGenerate(RuntimeException fault)
+ {
+ failGenerateAfter(0, fault);
+ }
+
+ void failGenerateAfter(int successfulGenerates, RuntimeException fault)
+ {
+ generatesBeforeFault.set(successfulGenerates);
+ generateFault = fault;
}
- public String getValue()
+ void healGenerate()
{
- return value;
+ generateFault = null;
}
- public void waitForValueToBeGotten(long length, TimeUnit timeUnit) throws
InterruptedException
+ void generateNull()
{
- gotValueLatch.await(length, timeUnit);
+ generateNull = true;
}
}
}
diff --git
a/server/src/main/java/org/apache/druid/guice/http/DruidHttpClientConfig.java
b/server/src/main/java/org/apache/druid/guice/http/DruidHttpClientConfig.java
index e180a968f1f..a9d4847d4f0 100644
---
a/server/src/main/java/org/apache/druid/guice/http/DruidHttpClientConfig.java
+++
b/server/src/main/java/org/apache/druid/guice/http/DruidHttpClientConfig.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.Min;
import org.apache.druid.java.util.common.HumanReadableBytes;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
import org.apache.druid.utils.JvmUtils;
import org.joda.time.Duration;
import org.joda.time.Period;
@@ -40,7 +41,7 @@ public class DruidHttpClientConfig
private static final Logger LOG = new Logger(DruidHttpClientConfig.class);
@JsonProperty
- @Min(0)
+ @Min(1)
private int numConnections = 20;
@JsonProperty
@@ -73,6 +74,9 @@ public class DruidHttpClientConfig
@JsonProperty
private Boolean eagerInitialization = null;
+ @JsonProperty
+ private ResourcePool.Implementation poolImplementation =
ResourcePool.Implementation.ADAPTIVE;
+
@JsonProperty
private long clientConnectTimeout = TimeUnit.MILLISECONDS.toMillis(500);
@@ -150,6 +154,11 @@ public class DruidHttpClientConfig
return eagerInitialization;
}
+ public ResourcePool.Implementation getPoolImplementation()
+ {
+ return poolImplementation;
+ }
+
public long getClientConnectTimeout()
{
return clientConnectTimeout;
diff --git
a/server/src/main/java/org/apache/druid/guice/http/HttpClientModule.java
b/server/src/main/java/org/apache/druid/guice/http/HttpClientModule.java
index ef4e1ad1d9f..d4c7f318f6b 100644
--- a/server/src/main/java/org/apache/druid/guice/http/HttpClientModule.java
+++ b/server/src/main/java/org/apache/druid/guice/http/HttpClientModule.java
@@ -42,9 +42,11 @@ import
org.apache.druid.java.util.http.client.AbstractHttpClient;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.HttpClientConfig;
import org.apache.druid.java.util.http.client.HttpClientInit;
+import org.apache.druid.java.util.http.client.NettyHttpClient;
import org.apache.druid.java.util.http.client.Request;
import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.metrics.HttpClientPoolRegistry;
import org.apache.druid.server.security.Escalator;
import org.joda.time.Duration;
@@ -96,23 +98,27 @@ public class HttpClientModule implements Module
public static class HttpClientProvider extends
AbstractHttpClientProvider<HttpClient>
{
+ private final Class<? extends Annotation> annotationClazz;
private final boolean isEscalated;
private final boolean eagerByDefault;
private Escalator escalator;
private DruidNode node;
+ private HttpClientPoolRegistry poolRegistry;
public HttpClientProvider(Class<? extends Annotation> annotationClazz,
boolean isEscalated, boolean eagerByDefault)
{
super(annotationClazz);
+ this.annotationClazz = annotationClazz;
this.isEscalated = isEscalated;
this.eagerByDefault = eagerByDefault;
}
@Inject
- public void inject(Escalator escalator, @Self DruidNode node)
+ public void inject(Escalator escalator, @Self DruidNode node,
HttpClientPoolRegistry poolRegistry)
{
this.escalator = escalator;
this.node = node;
+ this.poolRegistry = poolRegistry;
}
@Override
@@ -124,6 +130,7 @@ public class HttpClientModule implements Module
.builder()
.withNumConnections(config.getNumConnections())
.withEagerInitialization(config.isEagerInitialization(eagerByDefault))
+ .withPoolImplementation(config.getPoolImplementation())
.withReadTimeout(config.getReadTimeout())
.withConnectTimeout(config.getConnectTimeout())
.withByteBufAllocator(resolveAllocator(config.getAllocator()))
@@ -139,10 +146,11 @@ public class HttpClientModule implements Module
builder.withSslContext(sslContextBinding.getProvider().get());
}
- HttpClient client = HttpClientInit.createClient(
+ NettyHttpClient client = HttpClientInit.createNettyClient(
builder.build(),
getLifecycleProvider().get()
);
+ poolRegistry.register(clientName(), client.getPool());
HttpClient clientWithUserAgent = new AbstractHttpClient()
{
@Override
@@ -163,6 +171,16 @@ public class HttpClientModule implements Module
return clientWithUserAgent;
}
}
+
+ /**
+ * The binding annotation of this client, which is what tells its pool
apart from the pools of the other clients
+ * of the same process - {@code druid.global.http} backs two of them.
+ */
+ private String clientName()
+ {
+ final String simpleName = annotationClazz.getSimpleName();
+ return StringUtils.toLowerCase(simpleName.substring(0, 1)) +
simpleName.substring(1);
+ }
}
/**
diff --git
a/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolMonitor.java
b/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolMonitor.java
new file mode 100644
index 00000000000..49007fdc93c
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolMonitor.java
@@ -0,0 +1,91 @@
+/*
+ * 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.druid.server.metrics;
+
+import com.google.inject.Inject;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
+import org.apache.druid.java.util.metrics.AbstractMonitor;
+
+import java.util.Map;
+
+/**
+ * Emits one row per remote end that an HTTP client of this process pools
connections for: what happened since the
+ * previous emission, plus what that remote end holds right now. Aggregating
the rows of a client is left to whoever
+ * receives them.
+ *
+ * This monitor drains the counters of the pools it reads, so it must be the
only reader of them.
+ */
+public class HttpClientPoolMonitor extends AbstractMonitor
+{
+ public static final String CLIENT_DIMENSION = "httpClient";
+ public static final String SERVER_DIMENSION = "server";
+
+ public static final String OPENED_METRIC = "httpClient/pool/opened";
+ public static final String CLOSED_METRIC = "httpClient/pool/closed";
+ public static final String ERRORED_METRIC = "httpClient/pool/errored";
+ public static final String TIMED_OUT_METRIC = "httpClient/pool/timedOut";
+ public static final String TAKEN_METRIC = "httpClient/pool/taken";
+ public static final String RETURNED_METRIC = "httpClient/pool/returned";
+ public static final String USED_METRIC = "httpClient/pool/used";
+ public static final String IDLE_METRIC = "httpClient/pool/idle";
+
+ private final HttpClientPoolRegistry registry;
+
+ @Inject
+ public HttpClientPoolMonitor(HttpClientPoolRegistry registry)
+ {
+ this.registry = registry;
+ }
+
+ @Override
+ public boolean doMonitor(ServiceEmitter emitter)
+ {
+ for (Map.Entry<String, ResourcePool<?, ?>> client :
registry.getPools().entrySet()) {
+ for (Map.Entry<?, ResourcePool.Stats> pool :
client.getValue().drainStats().entrySet()) {
+ final ResourcePool.Stats stats = pool.getValue();
+ final ServiceMetricEvent.Builder builder = ServiceMetricEvent.builder()
+ .setDimension(CLIENT_DIMENSION, client.getKey())
+ .setDimension(SERVER_DIMENSION, serverOf(pool.getKey()));
+ emitter.emit(builder.setMetric(OPENED_METRIC, stats.opened()));
+ emitter.emit(builder.setMetric(CLOSED_METRIC, stats.closed()));
+ emitter.emit(builder.setMetric(ERRORED_METRIC, stats.errored()));
+ emitter.emit(builder.setMetric(TIMED_OUT_METRIC, stats.timedOut()));
+ emitter.emit(builder.setMetric(TAKEN_METRIC, stats.taken()));
+ emitter.emit(builder.setMetric(RETURNED_METRIC, stats.returned()));
+ emitter.emit(builder.setMetric(USED_METRIC, stats.used()));
+ emitter.emit(builder.setMetric(IDLE_METRIC, stats.idle()));
+ }
+ }
+ return true;
+ }
+
+ /**
+ * The pool key without its scheme, so that {@code host:port} reads the same
here as it does under the
+ * {@code server} dimension of the query metrics.
+ */
+ private static String serverOf(Object poolKey)
+ {
+ final String key = String.valueOf(poolKey);
+ final int schemeEnd = key.indexOf("://");
+ return schemeEnd < 0 ? key : key.substring(schemeEnd + 3);
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolRegistry.java
b/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolRegistry.java
new file mode 100644
index 00000000000..b0b213c0201
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/metrics/HttpClientPoolRegistry.java
@@ -0,0 +1,56 @@
+/*
+ * 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.druid.server.metrics;
+
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * The connection pool of every HTTP client of this process, under the name of
the client that owns it. A process runs
+ * several clients, so {@link HttpClientPoolMonitor} needs the name to keep
their metrics apart.
+ */
+@LazySingleton
+public class HttpClientPoolRegistry
+{
+ private static final Logger log = new Logger(HttpClientPoolRegistry.class);
+
+ private final Map<String, ResourcePool<?, ?>> pools = new
ConcurrentHashMap<>();
+
+ /**
+ * Registers the pool of a client, replacing the pool registered under that
name before, whose metrics then stop
+ * being reported.
+ */
+ public void register(String clientName, ResourcePool<?, ?> pool)
+ {
+ if (pools.put(clientName, pool) != null) {
+ log.warn("Client[%s] was already registered, only the pool registered
last is reported.", clientName);
+ }
+ }
+
+ public Map<String, ResourcePool<?, ?>> getPools()
+ {
+ return Collections.unmodifiableMap(pools);
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/guice/http/DruidHttpClientConfigTest.java
b/server/src/test/java/org/apache/druid/guice/http/DruidHttpClientConfigTest.java
new file mode 100644
index 00000000000..70b0dec5b5a
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/guice/http/DruidHttpClientConfigTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.druid.guice.http;
+
+import jakarta.validation.Validation;
+import org.apache.druid.guice.JsonConfigProvider;
+import org.apache.druid.guice.JsonConfigurator;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+/**
+ * Covers reading the connection pool knobs out of runtime.properties, which
is how an operator moves off the default
+ * pooling implementation or turns strict connection validation on.
+ */
+public class DruidHttpClientConfigTest
+{
+ private static final String PROPERTY_BASE = "druid.broker.http";
+
+ @Test
+ public void testPoolImplementationDefaultsToAdaptive()
+ {
+ Assertions.assertEquals(ResourcePool.Implementation.ADAPTIVE,
configure(new Properties()).getPoolImplementation());
+ }
+
+ @Test
+ public void testPoolImplementationCanBeSwitchedBackToRetaining()
+ {
+ final Properties properties = new Properties();
+ properties.setProperty(PROPERTY_BASE + ".poolImplementation", "retaining");
+
+ Assertions.assertEquals(ResourcePool.Implementation.RETAINING,
configure(properties).getPoolImplementation());
+ }
+
+ /**
+ * Either casing is accepted, since the constant is upper case while the
value serialises lower case.
+ */
+ @Test
+ public void testPoolImplementationIsNotCaseSensitive()
+ {
+ final Properties properties = new Properties();
+ properties.setProperty(PROPERTY_BASE + ".poolImplementation", "RETAINING");
+
+ Assertions.assertEquals(ResourcePool.Implementation.RETAINING,
configure(properties).getPoolImplementation());
+ }
+
+ /**
+ * A misspelled implementation fails the service at startup rather than
silently leaving the default in place.
+ */
+ @Test
+ public void testUnknownPoolImplementationIsRejected()
+ {
+ final Properties properties = new Properties();
+ properties.setProperty(PROPERTY_BASE + ".poolImplementation", "invalid");
+
+ Assertions.assertThrows(RuntimeException.class, () ->
configure(properties));
+ }
+
+ private static DruidHttpClientConfig configure(Properties properties)
+ {
+ final JsonConfigurator configurator = new JsonConfigurator(
+ new DefaultObjectMapper(),
+ Validation.buildDefaultValidatorFactory().getValidator()
+ );
+ final JsonConfigProvider<DruidHttpClientConfig> provider =
+ JsonConfigProvider.of(PROPERTY_BASE, DruidHttpClientConfig.class);
+ provider.inject(properties, configurator);
+ return provider.get();
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/guice/http/HttpClientPoolRegistrationTest.java
b/server/src/test/java/org/apache/druid/guice/http/HttpClientPoolRegistrationTest.java
new file mode 100644
index 00000000000..1ec9da78606
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/guice/http/HttpClientPoolRegistrationTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.druid.guice.http;
+
+import com.google.inject.Injector;
+import com.google.inject.Key;
+import com.google.inject.Module;
+import org.apache.druid.guice.GuiceInjectors;
+import org.apache.druid.guice.annotations.EscalatedGlobal;
+import org.apache.druid.guice.annotations.Global;
+import org.apache.druid.guice.annotations.Self;
+import org.apache.druid.initialization.Initialization;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.metrics.HttpClientPoolRegistry;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.server.security.NoopEscalator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Covers the wiring that lets {@code HttpClientPoolMonitor} find the
connection pools: every client of the process
+ * hands its pool to the registry, under a name that is its own.
+ */
+public class HttpClientPoolRegistrationTest
+{
+ @Test
+ public void testEachClientRegistersItsPoolUnderItsOwnName()
+ {
+ final Module testBindings = binder -> {
+ binder.bind(Escalator.class).toInstance(NoopEscalator.getInstance());
+ binder.bind(DruidNode.class)
+ .annotatedWith(Self.class)
+ .toInstance(new DruidNode("test", "localhost", false, 8080, null,
true, false));
+ };
+ final Injector injector = Initialization.makeInjectorWithModules(
+ GuiceInjectors.makeStartupInjector(),
+ List.of(testBindings, HttpClientModule.global(),
HttpClientModule.escalatedGlobal())
+ );
+
+ injector.getInstance(Key.get(HttpClient.class, Global.class));
+ injector.getInstance(Key.get(HttpClient.class, EscalatedGlobal.class));
+
+ Assertions.assertEquals(
+ Set.of("global", "escalatedGlobal"),
+ injector.getInstance(HttpClientPoolRegistry.class).getPools().keySet()
+ );
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/server/metrics/HttpClientPoolMonitorTest.java
b/server/src/test/java/org/apache/druid/server/metrics/HttpClientPoolMonitorTest.java
new file mode 100644
index 00000000000..8f28b8d6989
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/metrics/HttpClientPoolMonitorTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.druid.server.metrics;
+
+import org.apache.druid.java.util.http.client.pool.ResourceContainer;
+import org.apache.druid.java.util.http.client.pool.ResourceFactory;
+import org.apache.druid.java.util.http.client.pool.ResourcePool;
+import org.apache.druid.java.util.http.client.pool.ResourcePoolConfig;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class HttpClientPoolMonitorTest
+{
+ private static final String HISTORICAL1_URL =
"https://historical1.example.com:8283";;
+ private static final String HISTORICAL2_URL =
"https://historical2.example.com:8283";;
+ private static final Map<String, Object> HISTORICAL1 =
dimensions("historical1.example.com:8283");
+ private static final Map<String, Object> HISTORICAL2 =
dimensions("historical2.example.com:8283");
+
+ private HttpClientPoolRegistry registry;
+ private StubServiceEmitter emitter;
+ private ResourcePool<String, String> pool;
+
+ @BeforeEach
+ public void setUp()
+ {
+ registry = new HttpClientPoolRegistry();
+ emitter = new StubServiceEmitter("service", "host");
+ pool = new ResourcePool<>(
+ new TestResourceFactory(),
+ new ResourcePoolConfig(2, TimeUnit.MINUTES.toMillis(5)),
+ false
+ );
+ }
+
+ /**
+ * Every remote end gets its own row, tagged with the client that talks to
it, so that nothing has to be summed up
+ * before it is emitted.
+ */
+ @Test
+ public void testEveryRemoteEndIsReportedOnItsOwn()
+ {
+ registry.register("global", pool);
+ pool.take(HISTORICAL1_URL).returnResource();
+ pool.take(HISTORICAL2_URL).returnResource();
+ pool.take(HISTORICAL2_URL).returnResource();
+
+ new HttpClientPoolMonitor(registry).doMonitor(emitter);
+
+ Assertions.assertEquals(List.of(1L),
emitter.getMetricValues(HttpClientPoolMonitor.TAKEN_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(2L),
emitter.getMetricValues(HttpClientPoolMonitor.TAKEN_METRIC, HISTORICAL2));
+ Assertions.assertEquals(List.of(1L),
emitter.getMetricValues(HttpClientPoolMonitor.OPENED_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(1),
emitter.getMetricValues(HttpClientPoolMonitor.IDLE_METRIC, HISTORICAL2));
+ }
+
+ /**
+ * The counted events are drained as they are emitted, so a quiet remote end
reports zeroes rather than the totals
+ * of the previous emission, while what it holds keeps being reported.
+ */
+ @Test
+ public void testEmissionsReportWhatHappenedSinceThePreviousOne()
+ {
+ registry.register("global", pool);
+ final HttpClientPoolMonitor monitor = new HttpClientPoolMonitor(registry);
+
+ pool.take(HISTORICAL1_URL).returnResource();
+ monitor.doMonitor(emitter);
+ emitter.flush();
+
+ monitor.doMonitor(emitter);
+ Assertions.assertEquals(List.of(0L),
emitter.getMetricValues(HttpClientPoolMonitor.OPENED_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(0L),
emitter.getMetricValues(HttpClientPoolMonitor.TAKEN_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(1),
emitter.getMetricValues(HttpClientPoolMonitor.IDLE_METRIC, HISTORICAL1));
+ }
+
+ /**
+ * The connections in the hands of callers are reported while they are held,
which is what the load over one remote
+ * end looks like, and go back to zero once they are given back.
+ */
+ @Test
+ public void testUsedConnectionsAreReportedWhileTheyAreHeld()
+ {
+ registry.register("global", pool);
+ final HttpClientPoolMonitor monitor = new HttpClientPoolMonitor(registry);
+
+ final ResourceContainer<String> lent = pool.take(HISTORICAL1_URL);
+ monitor.doMonitor(emitter);
+ Assertions.assertEquals(List.of(1),
emitter.getMetricValues(HttpClientPoolMonitor.USED_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(0),
emitter.getMetricValues(HttpClientPoolMonitor.IDLE_METRIC, HISTORICAL1));
+
+ lent.returnResource();
+ emitter.flush();
+ monitor.doMonitor(emitter);
+ Assertions.assertEquals(List.of(1L),
emitter.getMetricValues(HttpClientPoolMonitor.RETURNED_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(0),
emitter.getMetricValues(HttpClientPoolMonitor.USED_METRIC, HISTORICAL1));
+ Assertions.assertEquals(List.of(1),
emitter.getMetricValues(HttpClientPoolMonitor.IDLE_METRIC, HISTORICAL1));
+ }
+
+ /**
+ * The scheme of the pool key is left out, so that the remote end reads the
same as under the {@code server}
+ * dimension of the query metrics.
+ */
+ @Test
+ public void testTheRemoteEndIsReportedWithoutItsScheme()
+ {
+ registry.register("global", pool);
+ pool.take("https://1.2.3.4:8283";).returnResource();
+
+ new HttpClientPoolMonitor(registry).doMonitor(emitter);
+
+ Assertions.assertEquals(
+ List.of(1L),
+ emitter.getMetricValues(HttpClientPoolMonitor.TAKEN_METRIC,
dimensions("1.2.3.4:8283"))
+ );
+ }
+
+ /**
+ * A client name registered twice reports the pool registered last, rather
than failing the process it was only
+ * meant to report on.
+ */
+ @Test
+ public void testAReRegisteredClientNameReportsTheLastPool()
+ {
+ final ResourcePool<String, String> replacement = new ResourcePool<>(
+ new TestResourceFactory(),
+ new ResourcePoolConfig(2, TimeUnit.MINUTES.toMillis(5)),
+ false
+ );
+ registry.register("global", pool);
+ registry.register("global", replacement);
+
+ Assertions.assertEquals(Map.of("global", replacement),
registry.getPools());
+ }
+
+ @Test
+ public void testNothingIsEmittedWithoutARegisteredPool()
+ {
+ new HttpClientPoolMonitor(registry).doMonitor(emitter);
+ Assertions.assertEquals(List.of(),
emitter.getMetricEvents(HttpClientPoolMonitor.OPENED_METRIC));
+ }
+
+ /**
+ * A registered client that has not talked to anybody yet has no key to
report.
+ */
+ @Test
+ public void testNothingIsEmittedForAnUntouchedPool()
+ {
+ registry.register("global", pool);
+ new HttpClientPoolMonitor(registry).doMonitor(emitter);
+ Assertions.assertEquals(List.of(),
emitter.getMetricEvents(HttpClientPoolMonitor.OPENED_METRIC));
+ }
+
+ private static Map<String, Object> dimensions(String server)
+ {
+ return Map.of(HttpClientPoolMonitor.CLIENT_DIMENSION, "global",
HttpClientPoolMonitor.SERVER_DIMENSION, server);
+ }
+
+ private static class TestResourceFactory implements ResourceFactory<String,
String>
+ {
+ private final AtomicInteger sequence = new AtomicInteger();
+
+ @Override
+ public String generate(String key)
+ {
+ return key + "#" + sequence.getAndIncrement();
+ }
+
+ @Override
+ public boolean isGood(String resource)
+ {
+ return true;
+ }
+
+ @Override
+ public void close(String resource)
+ {
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
