This is an automated email from the ASF dual-hosted git repository.
gian 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 ccb386e13d1 Make the HTTP client that lives beneath fabric8
KubernetesClient configurable (#18540)
ccb386e13d1 is described below
commit ccb386e13d11c7c733b380c94e2e48d49c3e95c5
Author: Lucas Capistrant <[email protected]>
AuthorDate: Wed Oct 15 07:16:44 2025 -0500
Make the HTTP client that lives beneath fabric8 KubernetesClient
configurable (#18540)
* Make the httpclient backing fabric8 KubernetesClient pluggable
* fix checkstyle
* fix licenses
* Cleanup prefixes for pluggable http client config
* Default okhttp and native jdk http client threadpools to static 20 threads
* experimental docs for http client config
* Cleanup docs and make native jdk client name more specific
* Fix unit tests
* fix dependency analyzer
* Make okhttp use the custom executor by default and bump its thread count
Also enforce proper setting of max threads for okhttp
* make native jdk http client configs more robust
* fix checkstyle
* Flip to okhttp as underlying http client for fabric8
* Revert "Flip to okhttp as underlying http client for fabric8"
This reverts commit 8b40ab7434c6a4f572d670b0d943dafb1c3a24fd.
* Turn off custom dispatcher for okhttp
* slim down docs for this experimental stuff and point to github issue
---
docs/development/extensions-core/k8s-jobs.md | 40 +++++
.../kubernetes-overlord-extensions/pom.xml | 21 ++-
.../k8s/overlord/KubernetesOverlordModule.java | 94 +++++++++++-
.../k8s/overlord/common/DruidKubernetesClient.java | 5 +-
.../DruidKubernetesHttpClientFactory.java} | 30 +---
.../jdk/DruidKubernetesJdkHttpClientConfig.java} | 33 +++--
.../jdk/DruidKubernetesJdkHttpClientFactory.java | 61 ++++++++
.../DruidKubernetesOkHttpHttpClientConfig.java} | 41 ++++--
.../DruidKubernetesOkHttpHttpClientFactory.java | 62 ++++++++
.../DruidKubernetesVertxHttpClientConfig.java} | 4 +-
.../DruidKubernetesVertxHttpClientFactory.java} | 13 +-
.../k8s/overlord/KubernetesOverlordModuleTest.java | 90 +++++++++++
.../overlord/KubernetesTaskRunnerFactoryTest.java | 5 +-
.../DruidKubernetesJdkHttpClientConfigTest.java | 156 ++++++++++++++++++++
.../DruidKubernetesOkHttpHttpClientConfigTest.java | 164 +++++++++++++++++++++
.../DruidPeonClientIntegrationTest.java | 5 +-
licenses.yaml | 2 +
website/.spelling | 4 +
18 files changed, 754 insertions(+), 76 deletions(-)
diff --git a/docs/development/extensions-core/k8s-jobs.md
b/docs/development/extensions-core/k8s-jobs.md
index 00b6024bb87..289934646af 100644
--- a/docs/development/extensions-core/k8s-jobs.md
+++ b/docs/development/extensions-core/k8s-jobs.md
@@ -856,3 +856,43 @@ To do this, set the following property.
|`druid.indexer.runner.k8sAndWorker.runnerStrategy.taskType.default`| `String`
(e.g., `k8s`, `worker`) | Specifies the default runner to use if no overrides
apply. This setting ensures there is always a fallback runner
available.|None|No|
|`druid.indexer.runner.k8sAndWorker.runnerStrategy.taskType.overrides`|
`JsonObject`(e.g., `{"index_kafka": "worker"}`)| Defines task-specific
overrides for runner types. Each entry sets a task type to a specific runner,
allowing fine control. |`{}`|No|
+### Experimental Fabric8 Http Client Configurations
+
+:::warning
+
+This section is experimental and subject to change. The Druid developer
community intends on selecting a stable default HTTP client and configuration
in the future, simplifying configuration and distribution packaging. This means
that not all exposed HTTP clients and their configurations will be supported
long term. If you opt in to using this experimental configuration, please
provide feedback to the Druid developer community ([GitHub
Issue](https://github.com/apache/druid/issues/18629) [...]
+
+:::
+
+The extension uses [fabric8
KubernetesClient](https://github.com/fabric8io/kubernetes-client) to
communicate with the Kubernetes API server. This client creates an
+underlying HTTP Client using a pluggable HTTP client library. By default, the
client is
[vert.x](https://github.com/fabric8io/kubernetes-client/tree/main/httpclient-vertx).
The legacy default
+was
[okhttp](https://github.com/fabric8io/kubernetes-client/tree/main/httpclient-okhttp).
+
+|Property| Possible Values |Description| Default |required|
+|--------|-----------------|-----------|---------|--------|
+|`druid.indexer.runner.k8sAndWorker.http.httpClientType`|`String` (e.g.,
`okhttp`, `vertx`, `javaStandardHttp`)|Specifies the HTTP client library to be
used by the worker task runner for communication with worker nodes.|`vertx`|No|
+
+#### vert.x HTTP Client
+
+|Property| Possible Values |Description| Default |required|
+|--------|-----------------|-----------|---------|--------|
+|`druid.indexer.runner.k8sAndWorker.http.vertx.workerPoolSize`|`Integer`|...|20|No|
+|`druid.indexer.runner.k8sAndWorker.http.vertx.eventLoopPoolSize`|`Integer`|...|`2
* number cores`|No|
+|`druid.indexer.runner.k8sAndWorker.http.vertx.internalBlockingPoolSize`|`Integer`|...|20|No|
+
+#### OkHttp Client
+
+|Property| Possible Values |Description| Default |required|
+|--------|-----------------|-----------|---------|--------|
+|`druid.indexer.runner.k8sAndWorker.http.okhttp.useCustomDispatcherExecutor`|`Boolean`|Flag
indicating if okhttp client will use a custom defined thread pool for okhttp
http client request dispatcher|false|No|
+|`druid.indexer.runner.k8sAndWorker.http.okhttp.coreWorkerThreads`|`Integer`|The
number of threads to keep in the pool, even if they are idle. Only applicable
if `useCustomDispatcherExecutor` is true.|50|No|
+|`druid.indexer.runner.k8sAndWorker.http.okhttp.maxWorkerThreads`|`Integer`|Maximum
number of threads in the custom thread pool for okhttp client request
dispatcher. Must be greater than or equal to
`druid.indexer.runner.k8sAndWorker.http.okhttp.coreWorkerThreads` Only
applicable if `useCustomDispatcherExecutor` is
true.|`druid.indexer.runner.k8sAndWorker.http.okhttp.coreWorkerThreads`|No|
+|`druid.indexer.runner.k8sAndWorker.http.okhttp.workerThreadKeepAliveTime`|`Long`|Idle
timeout in seconds for non-core threads in the worker thread pool. Only
applicable if `useCustomDispatcherExecutor` is true.|60|No|
+
+#### Native Java HTTP Client
+
+|Property| Possible Values |Description| Default |required|
+|--------|-----------------|-----------|---------|--------|
+|`druid.indexer.runner.k8sAndWorker.http.javaStandardHttp.coreWorkerThreads`|`Integer`|The
number of threads to keep in the pool, even if they are idle.|20|No|
+|`druid.indexer.runner.k8sAndWorker.http.javaStandardHttp.maxWorkerThreads`|`Integer`|Maximum
number of threads in the custom thread pool for okhttp client request
dispatcher.|20|No|
+|`druid.indexer.runner.k8sAndWorker.http.javaStandardHttp.workerThreadKeepAliveTime`|`Long`|Idle
timeout in seconds for non-core threads in the worker thread pool.|60|No|
diff --git a/extensions-core/kubernetes-overlord-extensions/pom.xml
b/extensions-core/kubernetes-overlord-extensions/pom.xml
index 63512884ba7..55de71c908d 100644
--- a/extensions-core/kubernetes-overlord-extensions/pom.xml
+++ b/extensions-core/kubernetes-overlord-extensions/pom.xml
@@ -121,18 +121,27 @@
<artifactId>kubernetes-client</artifactId>
<version>${fabric8.version}</version>
<scope>runtime</scope>
- <exclusions>
- <exclusion>
- <groupId>io.fabric8</groupId>
- <artifactId>kubernetes-httpclient-okhttp</artifactId>
- </exclusion>
- </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-httpclient-jdk</artifactId>
+ <version>${fabric8.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-httpclient-okhttp</artifactId>
+ <version>${fabric8.version}</version>
</dependency>
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-httpclient-vertx</artifactId>
<version>${fabric8.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>okhttp</artifactId>
+ <version>4.12.0</version>
+ </dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-core</artifactId>
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
index 55fb08627b2..021abef0828 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
@@ -54,7 +54,13 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.lifecycle.Lifecycle;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
-import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.jdk.DruidKubernetesJdkHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.jdk.DruidKubernetesJdkHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.okhttp.DruidKubernetesOkHttpHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.okhttp.DruidKubernetesOkHttpHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientFactory;
import
org.apache.druid.k8s.overlord.execution.KubernetesTaskExecutionConfigResource;
import
org.apache.druid.k8s.overlord.execution.KubernetesTaskRunnerDynamicConfig;
import org.apache.druid.k8s.overlord.runnerstrategy.RunnerStrategy;
@@ -79,7 +85,11 @@ public class KubernetesOverlordModule implements DruidModule
+
".k8sAndWorker";
private static final String RUNNERSTRATEGY_PROPERTIES_FORMAT_STRING =
K8SANDWORKER_PROPERTIES_PREFIX
+
".runnerStrategy.%s";
- private static final String HTTPCLIENT_PROPERITES_PREFIX =
K8SANDWORKER_PROPERTIES_PREFIX + ".http";
+ private static final String K8SANDWORKER_HTTPCLIENT_PROPERTIES_PREFIX =
K8SANDWORKER_PROPERTIES_PREFIX + ".http";
+ private static final String HTTPCLIENT_TYPE_PROPERTY =
K8SANDWORKER_HTTPCLIENT_PROPERTIES_PREFIX + ".httpClientType";
+ private static final String VERTX_HTTPCLIENT_PROPERITES_PREFIX =
K8SANDWORKER_HTTPCLIENT_PROPERTIES_PREFIX + "." +
DruidKubernetesVertxHttpClientFactory.TYPE_NAME;
+ private static final String OKHTTP_HTTPCLIENT_PROPERITES_PREFIX =
K8SANDWORKER_HTTPCLIENT_PROPERTIES_PREFIX + "." +
DruidKubernetesOkHttpHttpClientFactory.TYPE_NAME;
+ public static final String JDK_HTTPCLIENT_PROPERITES_PREFIX =
K8SANDWORKER_HTTPCLIENT_PROPERTIES_PREFIX + "." +
DruidKubernetesJdkHttpClientFactory.TYPE_NAME;
@Override
public void configure(Binder binder)
@@ -114,14 +124,37 @@ public class KubernetesOverlordModule implements
DruidModule
Jerseys.addResource(binder, KubernetesTaskExecutionConfigResource.class);
- JsonConfigProvider.bind(binder, HTTPCLIENT_PROPERITES_PREFIX,
DruidKubernetesHttpClientConfig.class);
+ PolyBind.createChoiceWithDefault(
+ binder,
+ HTTPCLIENT_TYPE_PROPERTY,
+ Key.get(DruidKubernetesHttpClientFactory.class),
+ DruidKubernetesVertxHttpClientFactory.TYPE_NAME
+ );
+
+ final MapBinder<String, DruidKubernetesHttpClientFactory> factoryBinder =
+ PolyBind.optionBinder(binder,
Key.get(DruidKubernetesHttpClientFactory.class));
+
+ factoryBinder.addBinding(DruidKubernetesVertxHttpClientFactory.TYPE_NAME)
+ .toProvider(VertxHttpClientFactoryProvider.class)
+ .in(LazySingleton.class);
+
+ factoryBinder.addBinding(DruidKubernetesOkHttpHttpClientFactory.TYPE_NAME)
+ .toProvider(OkHttpHttpClientFactoryProvider.class)
+ .in(LazySingleton.class);
+ factoryBinder.addBinding(DruidKubernetesJdkHttpClientFactory.TYPE_NAME)
+ .toProvider(JdkHttpClientFactoryProvider.class)
+ .in(LazySingleton.class);
+
+ JsonConfigProvider.bind(binder, VERTX_HTTPCLIENT_PROPERITES_PREFIX,
DruidKubernetesVertxHttpClientConfig.class);
+ JsonConfigProvider.bind(binder, OKHTTP_HTTPCLIENT_PROPERITES_PREFIX,
DruidKubernetesOkHttpHttpClientConfig.class);
+ JsonConfigProvider.bind(binder, JDK_HTTPCLIENT_PROPERITES_PREFIX,
DruidKubernetesJdkHttpClientConfig.class);
}
@Provides
@LazySingleton
public DruidKubernetesClient makeKubernetesClient(
KubernetesTaskRunnerConfig kubernetesTaskRunnerConfig,
- DruidKubernetesHttpClientConfig httpClientConfig,
+ DruidKubernetesHttpClientFactory httpClientFactory,
Lifecycle lifecycle
)
{
@@ -133,7 +166,7 @@ public class KubernetesOverlordModule implements DruidModule
config.setHttpProxy(null);
}
- client = new DruidKubernetesClient(httpClientConfig, config);
+ client = new DruidKubernetesClient(httpClientFactory, config);
lifecycle.addHandler(
new Lifecycle.Handler()
@@ -280,4 +313,55 @@ public class KubernetesOverlordModule implements
DruidModule
return provider.get();
}
}
+
+ private static class VertxHttpClientFactoryProvider implements
Provider<DruidKubernetesHttpClientFactory>
+ {
+ private DruidKubernetesVertxHttpClientConfig config;
+
+ @Inject
+ public void inject(DruidKubernetesVertxHttpClientConfig config)
+ {
+ this.config = config; // Guice injects the Vertx-specific config
+ }
+
+ @Override
+ public DruidKubernetesHttpClientFactory get()
+ {
+ return new DruidKubernetesVertxHttpClientFactory(config);
+ }
+ }
+
+ private static class OkHttpHttpClientFactoryProvider implements
Provider<DruidKubernetesHttpClientFactory>
+ {
+ private DruidKubernetesOkHttpHttpClientConfig config;
+
+ @Inject
+ public void inject(DruidKubernetesOkHttpHttpClientConfig config)
+ {
+ this.config = config; // Guice injects the OkHttp-specific config
+ }
+
+ @Override
+ public DruidKubernetesHttpClientFactory get()
+ {
+ return new DruidKubernetesOkHttpHttpClientFactory(config);
+ }
+ }
+
+ private static class JdkHttpClientFactoryProvider implements
Provider<DruidKubernetesHttpClientFactory>
+ {
+ private DruidKubernetesJdkHttpClientConfig config;
+
+ @Inject
+ public void inject(DruidKubernetesJdkHttpClientConfig config)
+ {
+ this.config = config;
+ }
+
+ @Override
+ public DruidKubernetesHttpClientFactory get()
+ {
+ return new DruidKubernetesJdkHttpClientFactory(config);
+ }
+ }
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
index 9c1940bb1f4..86628673248 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
@@ -22,15 +22,16 @@ package org.apache.druid.k8s.overlord.common;
import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
public class DruidKubernetesClient implements KubernetesClientApi
{
private final KubernetesClient kubernetesClient;
- public DruidKubernetesClient(DruidKubernetesHttpClientConfig
httpClientConfig, Config kubernetesClientConfig)
+ public DruidKubernetesClient(DruidKubernetesHttpClientFactory
httpClientFactory, Config kubernetesClientConfig)
{
this.kubernetesClient = new KubernetesClientBuilder()
- .withHttpClientFactory(new
DruidKubernetesHttpClientFactory(httpClientConfig))
+ .withHttpClientFactory(httpClientFactory)
.withConfig(kubernetesClientConfig)
.build();
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/DruidKubernetesHttpClientFactory.java
similarity index 53%
copy from
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
copy to
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/DruidKubernetesHttpClientFactory.java
index b659f8e9af5..e0b803f3840 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/DruidKubernetesHttpClientFactory.java
@@ -17,34 +17,10 @@
* under the License.
*/
-package org.apache.druid.k8s.overlord.common;
+package org.apache.druid.k8s.overlord.common.httpclient;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import io.vertx.core.VertxOptions;
+import io.fabric8.kubernetes.client.http.HttpClient;
-public class DruidKubernetesHttpClientConfig
+public interface DruidKubernetesHttpClientFactory extends HttpClient.Factory
{
- @JsonProperty
- private int workerPoolSize = VertxOptions.DEFAULT_WORKER_POOL_SIZE;
-
- @JsonProperty
- private int eventLoopPoolSize = VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE;
-
- @JsonProperty
- private int internalBlockingPoolSize =
VertxOptions.DEFAULT_INTERNAL_BLOCKING_POOL_SIZE;
-
- public int getWorkerPoolSize()
- {
- return workerPoolSize;
- }
-
- public int getEventLoopPoolSize()
- {
- return eventLoopPoolSize;
- }
-
- public int getInternalBlockingPoolSize()
- {
- return internalBlockingPoolSize;
- }
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfig.java
similarity index 60%
copy from
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
copy to
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfig.java
index b659f8e9af5..49dd043308b 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfig.java
@@ -17,34 +17,43 @@
* under the License.
*/
-package org.apache.druid.k8s.overlord.common;
+package org.apache.druid.k8s.overlord.common.httpclient.jdk;
+
import com.fasterxml.jackson.annotation.JsonProperty;
-import io.vertx.core.VertxOptions;
-public class DruidKubernetesHttpClientConfig
+import javax.annotation.Nullable;
+
+public class DruidKubernetesJdkHttpClientConfig
{
+
@JsonProperty
- private int workerPoolSize = VertxOptions.DEFAULT_WORKER_POOL_SIZE;
+ @Nullable
+ private Integer maxWorkerThreads = null;
@JsonProperty
- private int eventLoopPoolSize = VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE;
+ private int coreWorkerThreads = 50;
@JsonProperty
- private int internalBlockingPoolSize =
VertxOptions.DEFAULT_INTERNAL_BLOCKING_POOL_SIZE;
+ private long workerThreadKeepAliveTime = 60L;
- public int getWorkerPoolSize()
+ public int getMaxWorkerThreads()
{
- return workerPoolSize;
+ if (maxWorkerThreads == null || maxWorkerThreads < coreWorkerThreads) {
+ return coreWorkerThreads;
+ } else {
+ return maxWorkerThreads;
+ }
}
- public int getEventLoopPoolSize()
+ public int getCoreWorkerThreads()
{
- return eventLoopPoolSize;
+ return coreWorkerThreads;
}
- public int getInternalBlockingPoolSize()
+ public long getWorkerThreadKeepAliveTime()
{
- return internalBlockingPoolSize;
+ return workerThreadKeepAliveTime;
}
+
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientFactory.java
new file mode 100644
index 00000000000..583d3692a54
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientFactory.java
@@ -0,0 +1,61 @@
+/*
+ * 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.k8s.overlord.common.httpclient.jdk;
+
+import io.fabric8.kubernetes.client.jdkhttp.JdkHttpClientFactory;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
+
+import java.net.http.HttpClient;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class DruidKubernetesJdkHttpClientFactory extends JdkHttpClientFactory
implements DruidKubernetesHttpClientFactory
+{
+ public static final String TYPE_NAME = "javaStandardHttp";
+ private final DruidKubernetesJdkHttpClientConfig config;
+
+ public
DruidKubernetesJdkHttpClientFactory(DruidKubernetesJdkHttpClientConfig config)
+ {
+ super();
+ this.config = config;
+ }
+
+ @Override
+ protected HttpClient.Builder createNewHttpClientBuilder()
+ {
+ ExecutorService executorService = new ThreadPoolExecutor(
+ config.getCoreWorkerThreads(),
+ config.getMaxWorkerThreads(),
+ config.getWorkerThreadKeepAliveTime(),
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ Execs.makeThreadFactory("JdkHttpClient-%d")
+ );
+ return HttpClient.newBuilder().executor(executorService);
+ }
+
+ @Override
+ protected void additionalConfig(HttpClient.Builder builder)
+ {
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfig.java
similarity index 52%
copy from
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
copy to
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfig.java
index b659f8e9af5..e5af2a4264f 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfig.java
@@ -17,34 +17,51 @@
* under the License.
*/
-package org.apache.druid.k8s.overlord.common;
+package org.apache.druid.k8s.overlord.common.httpclient.okhttp;
import com.fasterxml.jackson.annotation.JsonProperty;
-import io.vertx.core.VertxOptions;
-public class DruidKubernetesHttpClientConfig
+import javax.annotation.Nullable;
+import javax.validation.constraints.Min;
+
+public class DruidKubernetesOkHttpHttpClientConfig
{
@JsonProperty
- private int workerPoolSize = VertxOptions.DEFAULT_WORKER_POOL_SIZE;
+ private boolean useCustomDispatcherExecutor = false;
+
+ @JsonProperty
+ @Nullable
+ private Integer maxWorkerThreads = null;
@JsonProperty
- private int eventLoopPoolSize = VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE;
+ @Min(1)
+ private int coreWorkerThreads = 50;
@JsonProperty
- private int internalBlockingPoolSize =
VertxOptions.DEFAULT_INTERNAL_BLOCKING_POOL_SIZE;
+ private long workerThreadKeepAliveTime = 60L;
- public int getWorkerPoolSize()
+ public boolean isUseCustomDispatcherExecutor()
{
- return workerPoolSize;
+ return useCustomDispatcherExecutor;
}
- public int getEventLoopPoolSize()
+ public int getMaxWorkerThreads()
{
- return eventLoopPoolSize;
+ if (maxWorkerThreads == null || maxWorkerThreads < coreWorkerThreads) {
+ return coreWorkerThreads;
+ } else {
+ return maxWorkerThreads;
+ }
}
- public int getInternalBlockingPoolSize()
+ public int getCoreWorkerThreads()
{
- return internalBlockingPoolSize;
+ return coreWorkerThreads;
}
+
+ public long getWorkerThreadKeepAliveTime()
+ {
+ return workerThreadKeepAliveTime;
+ }
+
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientFactory.java
new file mode 100644
index 00000000000..7081302f4bd
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientFactory.java
@@ -0,0 +1,62 @@
+/*
+ * 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.k8s.overlord.common.httpclient.okhttp;
+
+import io.fabric8.kubernetes.client.okhttp.OkHttpClientFactory;
+import okhttp3.Dispatcher;
+import okhttp3.OkHttpClient;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class DruidKubernetesOkHttpHttpClientFactory extends
OkHttpClientFactory implements DruidKubernetesHttpClientFactory
+{
+ public static final String TYPE_NAME = "okhttp";
+ private final DruidKubernetesOkHttpHttpClientConfig config;
+
+ public
DruidKubernetesOkHttpHttpClientFactory(DruidKubernetesOkHttpHttpClientConfig
config)
+ {
+ super();
+ this.config = config;
+ }
+
+ @Override
+ protected void additionalConfig(OkHttpClient.Builder builder)
+ {
+ if (config.isUseCustomDispatcherExecutor()) {
+ ExecutorService executorService = new ThreadPoolExecutor(
+ config.getCoreWorkerThreads(),
+ config.getMaxWorkerThreads(),
+ config.getWorkerThreadKeepAliveTime(),
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ Execs.makeThreadFactory("OkHttpHttpClient-%d")
+ );
+ Dispatcher dispatcher = new Dispatcher(executorService);
+ dispatcher.setMaxRequests(config.getMaxWorkerThreads());
+ dispatcher.setMaxRequestsPerHost(config.getMaxWorkerThreads());
+ builder.dispatcher(dispatcher);
+ }
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientConfig.java
similarity index 92%
rename from
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
rename to
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientConfig.java
index b659f8e9af5..14e7a122d40 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientConfig.java
@@ -17,12 +17,12 @@
* under the License.
*/
-package org.apache.druid.k8s.overlord.common;
+package org.apache.druid.k8s.overlord.common.httpclient.vertx;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.vertx.core.VertxOptions;
-public class DruidKubernetesHttpClientConfig
+public class DruidKubernetesVertxHttpClientConfig
{
@JsonProperty
private int workerPoolSize = VertxOptions.DEFAULT_WORKER_POOL_SIZE;
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientFactory.java
similarity index 84%
rename from
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
rename to
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientFactory.java
index 1924eb11d81..f26c9ddcc1c 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/httpclient/vertx/DruidKubernetesVertxHttpClientFactory.java
@@ -17,30 +17,31 @@
* under the License.
*/
-package org.apache.druid.k8s.overlord.common;
+package org.apache.druid.k8s.overlord.common.httpclient.vertx;
-import io.fabric8.kubernetes.client.http.HttpClient;
import io.fabric8.kubernetes.client.vertx.VertxHttpClientBuilder;
import io.fabric8.kubernetes.client.vertx.VertxHttpClientFactory;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.file.FileSystemOptions;
import io.vertx.core.spi.resolver.ResolverProvider;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
/**
* Similar to {@link VertxHttpClientFactory} but allows us to override thread
pool configurations.
*/
-public class DruidKubernetesHttpClientFactory implements HttpClient.Factory
+public class DruidKubernetesVertxHttpClientFactory implements
DruidKubernetesHttpClientFactory
{
+ public static final String TYPE_NAME = "vertx";
private final Vertx vertx;
- public DruidKubernetesHttpClientFactory(final
DruidKubernetesHttpClientConfig httpClientConfig)
+ public DruidKubernetesVertxHttpClientFactory(final
DruidKubernetesVertxHttpClientConfig httpClientConfig)
{
this.vertx = createVertxInstance(httpClientConfig);
}
@Override
- public VertxHttpClientBuilder<DruidKubernetesHttpClientFactory> newBuilder()
+ public VertxHttpClientBuilder<DruidKubernetesVertxHttpClientFactory>
newBuilder()
{
return new VertxHttpClientBuilder<>(this, vertx);
}
@@ -49,7 +50,7 @@ public class DruidKubernetesHttpClientFactory implements
HttpClient.Factory
* Adapted from fabric8 kubernetes-client 7.1.0. We bring this here so we
can customize thread pool sizes
* and force usage of daemon threads.
*/
- private static Vertx createVertxInstance(final
DruidKubernetesHttpClientConfig httpClientConfig)
+ private static Vertx createVertxInstance(final
DruidKubernetesVertxHttpClientConfig httpClientConfig)
{
// fabric8 disables the async DNS resolver while creating Vertx.
// I'm not sure if we really need to do this, but I'm keeping it to align
behavior with upstream.
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesOverlordModuleTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesOverlordModuleTest.java
index 1d3f28b1aed..59c9508005f 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesOverlordModuleTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesOverlordModuleTest.java
@@ -39,6 +39,11 @@ import
org.apache.druid.indexing.overlord.hrtr.HttpRemoteTaskRunnerFactory;
import org.apache.druid.jackson.JacksonModule;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
+import
org.apache.druid.k8s.overlord.common.httpclient.DruidKubernetesHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.jdk.DruidKubernetesJdkHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.okhttp.DruidKubernetesOkHttpHttpClientFactory;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientFactory;
import org.apache.druid.k8s.overlord.taskadapter.MultiContainerTaskAdapter;
import org.apache.druid.k8s.overlord.taskadapter.PodTemplateTaskAdapter;
import org.apache.druid.k8s.overlord.taskadapter.SingleContainerTaskAdapter;
@@ -200,6 +205,91 @@ public class KubernetesOverlordModuleTest
Assert.assertTrue(adapter instanceof PodTemplateTaskAdapter);
}
+ @Test
+ public void test_httpClientFactory_defaultsToVertx()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ // Don't set httpClientType - should default to vertx
+
+ injector = makeInjectorWithProperties(props, false, true);
+ DruidKubernetesHttpClientFactory factory =
injector.getInstance(DruidKubernetesHttpClientFactory.class);
+
+ Assert.assertNotNull(factory);
+ Assert.assertTrue("Should default to Vertx HTTP client",
+ factory instanceof DruidKubernetesVertxHttpClientFactory);
+ }
+
+ @Test
+ public void test_httpClientFactory_okhttpSelection()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ props.setProperty("druid.indexer.runner.k8sAndWorker.http.httpClientType",
"okhttp");
+
+ injector = makeInjectorWithProperties(props, false, true);
+ DruidKubernetesHttpClientFactory factory =
injector.getInstance(DruidKubernetesHttpClientFactory.class);
+
+ Assert.assertNotNull(factory);
+ Assert.assertTrue("Should select OkHttp HTTP client",
+ factory instanceof
DruidKubernetesOkHttpHttpClientFactory);
+ }
+
+ @Test
+ public void test_httpClientFactory_vertxExplicitSelection()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ props.setProperty("druid.indexer.runner.k8sAndWorker.http.httpClientType",
"vertx");
+
+ injector = makeInjectorWithProperties(props, false, true);
+ DruidKubernetesHttpClientFactory factory =
injector.getInstance(DruidKubernetesHttpClientFactory.class);
+
+ Assert.assertNotNull(factory);
+ Assert.assertTrue("Should explicitly select Vertx HTTP client",
+ factory instanceof DruidKubernetesVertxHttpClientFactory);
+ }
+
+ @Test
+ public void test_httpClientFactory_jdkSelection()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ props.setProperty("druid.indexer.runner.k8sAndWorker.http.httpClientType",
"javaStandardHttp");
+
+ injector = makeInjectorWithProperties(props, false, true);
+ DruidKubernetesHttpClientFactory factory =
injector.getInstance(DruidKubernetesHttpClientFactory.class);
+
+ Assert.assertNotNull(factory);
+ Assert.assertTrue("Should select JDK HTTP client",
+ factory instanceof DruidKubernetesJdkHttpClientFactory);
+ }
+
+ @Test(expected = ProvisionException.class)
+ public void test_httpClientFactory_invalidTypeThrowsException()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ props.setProperty("druid.indexer.runner.k8sAndWorker.http.httpClientType",
"invalid");
+
+ injector = makeInjectorWithProperties(props, false, true);
+ injector.getInstance(DruidKubernetesHttpClientFactory.class);
+ }
+
+ @Test
+ public void test_druidKubernetesClient_createdWithVertxClient()
+ {
+ Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.namespace", "NAMESPACE");
+ // Don't set httpClientType - should default to vertx
+
+ injector = makeInjectorWithProperties(props, false, true);
+ DruidKubernetesClient client =
injector.getInstance(DruidKubernetesClient.class);
+
+ Assert.assertNotNull("DruidKubernetesClient should be created
successfully", client);
+ Assert.assertNotNull("Underlying Kubernetes client should be created",
client.getClient());
+ }
+
private Injector makeInjectorWithProperties(
final Properties props,
boolean isWorkerTypeRemote,
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
index 49720642216..6849f9a0ecc 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
@@ -26,8 +26,9 @@ import org.apache.druid.indexing.common.TestUtils;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
-import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
import org.apache.druid.k8s.overlord.common.K8sTaskId;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientFactory;
import org.apache.druid.k8s.overlord.taskadapter.TaskAdapter;
import org.apache.druid.tasklogs.NoopTaskLogs;
import org.apache.druid.tasklogs.TaskLogs;
@@ -57,7 +58,7 @@ public class KubernetesTaskRunnerFactoryTest
.build();
taskLogs = new NoopTaskLogs();
druidKubernetesClient =
- new DruidKubernetesClient(new DruidKubernetesHttpClientConfig(), new
ConfigBuilder().build());
+ new DruidKubernetesClient(new
DruidKubernetesVertxHttpClientFactory(new
DruidKubernetesVertxHttpClientConfig()), new ConfigBuilder().build());
taskAdapter = new TestTaskAdapter();
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfigTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfigTest.java
new file mode 100644
index 00000000000..0541d6a9ed3
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/jdk/DruidKubernetesJdkHttpClientConfigTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.k8s.overlord.common.httpclient.jdk;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class DruidKubernetesJdkHttpClientConfigTest
+{
+ private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+
+ @Test
+ public void testSerdeWithDefaults() throws IOException
+ {
+ String json = "{}";
+ DruidKubernetesJdkHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(50, config.getCoreWorkerThreads());
+ Assert.assertEquals(50, config.getMaxWorkerThreads());
+ Assert.assertEquals(60L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesJdkHttpClientConfig deserialized = OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testSerdeWithAllFieldsSet() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": 80,\n"
+ + " \"coreWorkerThreads\": 30,\n"
+ + " \"workerThreadKeepAliveTime\": 120\n"
+ + "}";
+
+ DruidKubernetesJdkHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(80, config.getMaxWorkerThreads());
+ Assert.assertEquals(120L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesJdkHttpClientConfig deserialized = OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testSerdeWithNullMaxWorkerThreads() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": null,\n"
+ + " \"coreWorkerThreads\": 40,\n"
+ + " \"workerThreadKeepAliveTime\": 90\n"
+ + "}";
+
+ DruidKubernetesJdkHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(40, config.getCoreWorkerThreads());
+ Assert.assertEquals(40, config.getMaxWorkerThreads());
+ Assert.assertEquals(90L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesJdkHttpClientConfig deserialized = OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsLogic()
+ {
+ DruidKubernetesJdkHttpClientConfig config = new
DruidKubernetesJdkHttpClientConfig();
+
+ Assert.assertEquals(50, config.getMaxWorkerThreads());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsGreaterThanCore() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": 100,\n"
+ + " \"coreWorkerThreads\": 30\n"
+ + "}";
+
+ DruidKubernetesJdkHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(100, config.getMaxWorkerThreads());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsLessThanCore() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": 20,\n"
+ + " \"coreWorkerThreads\": 30\n"
+ + "}";
+
+ DruidKubernetesJdkHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesJdkHttpClientConfig.class
+ );
+
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(30, config.getMaxWorkerThreads());
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfigTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfigTest.java
new file mode 100644
index 00000000000..1baae0adad3
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/httpclient/okhttp/DruidKubernetesOkHttpHttpClientConfigTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.k8s.overlord.common.httpclient.okhttp;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class DruidKubernetesOkHttpHttpClientConfigTest
+{
+ private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+
+ @Test
+ public void testSerdeWithDefaults() throws IOException
+ {
+ String json = "{}";
+ DruidKubernetesOkHttpHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertFalse(config.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(50, config.getCoreWorkerThreads());
+ Assert.assertEquals(50, config.getMaxWorkerThreads());
+ Assert.assertEquals(60L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesOkHttpHttpClientConfig deserialized =
OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.isUseCustomDispatcherExecutor(),
deserialized.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testSerdeWithAllFieldsSet() throws IOException
+ {
+ String json = "{\n"
+ + " \"useCustomDispatcherExecutor\": false,\n"
+ + " \"maxWorkerThreads\": 99,\n"
+ + " \"coreWorkerThreads\": 30,\n"
+ + " \"workerThreadKeepAliveTime\": 120\n"
+ + "}";
+
+ DruidKubernetesOkHttpHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertFalse(config.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(99, config.getMaxWorkerThreads());
+ Assert.assertEquals(120L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesOkHttpHttpClientConfig deserialized =
OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.isUseCustomDispatcherExecutor(),
deserialized.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testSerdeWithNullMaxWorkerThreads() throws IOException
+ {
+ String json = "{\n"
+ + " \"useCustomDispatcherExecutor\": true,\n"
+ + " \"maxWorkerThreads\": null,\n"
+ + " \"coreWorkerThreads\": 40,\n"
+ + " \"workerThreadKeepAliveTime\": 90\n"
+ + "}";
+
+ DruidKubernetesOkHttpHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertTrue(config.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(40, config.getCoreWorkerThreads());
+ Assert.assertEquals(40, config.getMaxWorkerThreads());
+ Assert.assertEquals(90L, config.getWorkerThreadKeepAliveTime());
+
+ String serialized = OBJECT_MAPPER.writeValueAsString(config);
+ DruidKubernetesOkHttpHttpClientConfig deserialized =
OBJECT_MAPPER.readValue(
+ serialized,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertEquals(config.isUseCustomDispatcherExecutor(),
deserialized.isUseCustomDispatcherExecutor());
+ Assert.assertEquals(config.getCoreWorkerThreads(),
deserialized.getCoreWorkerThreads());
+ Assert.assertEquals(config.getMaxWorkerThreads(),
deserialized.getMaxWorkerThreads());
+ Assert.assertEquals(config.getWorkerThreadKeepAliveTime(),
deserialized.getWorkerThreadKeepAliveTime());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsLogic()
+ {
+ DruidKubernetesOkHttpHttpClientConfig config = new
DruidKubernetesOkHttpHttpClientConfig();
+
+ Assert.assertEquals(50, config.getMaxWorkerThreads());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsGreaterThanCore() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": 100,\n"
+ + " \"coreWorkerThreads\": 30\n"
+ + "}";
+
+ DruidKubernetesOkHttpHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(100, config.getMaxWorkerThreads());
+ }
+
+ @Test
+ public void testMaxWorkerThreadsLessThanCore() throws IOException
+ {
+ String json = "{\n"
+ + " \"maxWorkerThreads\": 20,\n"
+ + " \"coreWorkerThreads\": 30\n"
+ + "}";
+
+ DruidKubernetesOkHttpHttpClientConfig config = OBJECT_MAPPER.readValue(
+ json,
+ DruidKubernetesOkHttpHttpClientConfig.class
+ );
+
+ Assert.assertEquals(30, config.getCoreWorkerThreads());
+ Assert.assertEquals(30, config.getMaxWorkerThreads());
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
index 446c2502fa1..33f8349c619 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
@@ -35,7 +35,6 @@ import org.apache.druid.indexing.common.task.Task;
import
org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexTuningConfig;
import org.apache.druid.k8s.overlord.KubernetesTaskRunnerConfig;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
-import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
import org.apache.druid.k8s.overlord.common.JobResponse;
import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.common.K8sTestUtils;
@@ -43,6 +42,8 @@ import
org.apache.druid.k8s.overlord.common.KubernetesClientApi;
import org.apache.druid.k8s.overlord.common.KubernetesPeonClient;
import org.apache.druid.k8s.overlord.common.PeonCommandContext;
import org.apache.druid.k8s.overlord.common.PeonPhase;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientConfig;
+import
org.apache.druid.k8s.overlord.common.httpclient.vertx.DruidKubernetesVertxHttpClientFactory;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.log.StartupLoggingConfig;
import org.apache.druid.server.metrics.NoopServiceEmitter;
@@ -87,7 +88,7 @@ public class DruidPeonClientIntegrationTest
new NamedType(ParallelIndexTuningConfig.class, "index_parallel"),
new NamedType(IndexTask.IndexTuningConfig.class, "index")
);
- k8sClient = new DruidKubernetesClient(new
DruidKubernetesHttpClientConfig(), new ConfigBuilder().build());
+ k8sClient = new DruidKubernetesClient(new
DruidKubernetesVertxHttpClientFactory(new
DruidKubernetesVertxHttpClientConfig()), new ConfigBuilder().build());
peonClient = new KubernetesPeonClient(k8sClient, "default", false, new
NoopServiceEmitter());
druidNode = new DruidNode(
"test",
diff --git a/licenses.yaml b/licenses.yaml
index 0c5e6e3b1a0..8b25dfb2f9d 100644
--- a/licenses.yaml
+++ b/licenses.yaml
@@ -886,6 +886,8 @@ libraries:
- io.fabric8: kubernetes-model-scheduling
- io.fabric8: kubernetes-model-storageclass
- io.fabric8: kubernetes-httpclient-vertx
+ - io.fabric8: kubernetes-httpclient-jdk
+ - io.fabric8: kubernetes-httpclient-okhttp
---
name: vertx
diff --git a/website/.spelling b/website/.spelling
index a79ddea2f62..2eaa410bace 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -175,6 +175,7 @@ Kerberos
KeyStores
Kinesis
Kubernetes
+KubernetesClient
Lakehouse
LDAPS
LRU
@@ -343,6 +344,7 @@ endpointConfig
enum
expectedType
expr
+fabric8
failover
failovers
featureSpec
@@ -455,6 +457,7 @@ numerics
numShards
parameterize
objectGlob
+okhttp
parameterized
parse_json
parseable
@@ -624,6 +627,7 @@ vectorizeVirtualColumns
versioned
versioning
virtualColumns
+vert.x
w.r.t.
walkthrough
whitelist
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
