This is an automated email from the ASF dual-hosted git repository.
FrankChen021 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 46e8f93eb83 feat: Add advertisedPlaintextPort to DruidNode for sidecar
proxy support (#19598)
46e8f93eb83 is described below
commit 46e8f93eb83132fd516ff7eb208125ed00fc6c5c
Author: Sadananda Aithal <[email protected]>
AuthorDate: Thu Oct 1 18:58:13 2026 +0530
feat: Add advertisedPlaintextPort to DruidNode for sidecar proxy support
(#19598)
* Add advertisedPlaintextPort to DruidNode for sidecar proxy support
Allow Druid nodes to advertise a different plaintext port (e.g. an envoy
sidecar port) for peer discovery while binding Jetty to the actual
service port. This enables east-west mTLS via a sidecar proxy without
changing Druid's bind configuration.
The new `advertisedPlaintextPort` JSON property on DruidNode controls
the port advertised during peer discovery. When unset (or <= 0), it
falls back to `plaintextPort` preserving existing behaviour.
* Fix DruidNode constructor source compatibility
Restore the nine-argument DruidNode(..., labels) overload that was
dropped when advertisedPlaintextPort was added to the @JsonCreator
constructor, and update the new advertisedPlaintextPort tests to use
the ten-argument form. Without this the server module did not compile.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* Use advertisedPlaintextPort for all peer-facing paths and document it
Address review feedback on #19598:
- DruidNode.getPortToUse() (and so getUriToUse()) returns the advertised
port when TLS is disabled, matching getHostAndPortToUse(). This covers
basic-security cache notifications, catalog sync, the sys server
properties fetch, Consul registration, Router->Broker selection and
shuffle partition locations.
- BaseDynamicConfigSyncer builds broker locations via
ServiceLocation.fromDruidNode().
- ThreadingTaskRunner and SingleTaskBackgroundRunner report task
locations on the advertised port.
- ForkingTaskRunner forces druid.advertisedPlaintextPort=-1 for forked
peons, since they listen on their own dynamic ports and must not
inherit the Middle Manager's advertised port.
- Document druid.advertisedPlaintextPort next to druid.plaintextPort.
* Report advertised port in Kubernetes task pod locations
Task pods launched by the Kubernetes task runner are reached by pod IP
from KubernetesPeonLifecycle.getTaskLocation(), not service discovery,
so druid.advertisedPlaintextPort never applies to them and the location
always used DruidK8sConstants.PORT.
- Add druid.indexer.runner.advertisedPlaintextPort to the k8s and
multik8s runner configs. When positive, task locations report it
instead of the peon's own port, e.g. a sidecar proxy in the task pod.
- Document the property and how it relates to druid.advertisedPlaintextPort.
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
docs/configuration/index.md | 23 ++++
docs/development/extensions-core/k8s-jobs.md | 1 +
docs/querying/sql-metadata-tables.md | 2 +-
.../consul/discovery/ConsulServiceIdsTest.java | 6 +
.../k8s/overlord/KubernetesPeonLifecycle.java | 20 +++-
.../overlord/KubernetesPeonLifecycleFactory.java | 11 +-
.../k8s/overlord/KubernetesTaskRunnerConfig.java | 17 ++-
.../KubernetesTaskRunnerEffectiveConfig.java | 8 ++
.../k8s/overlord/KubernetesTaskRunnerFactory.java | 3 +-
.../overlord/KubernetesTaskRunnerStaticConfig.java | 17 ++-
.../MultipleKubernetesTaskRunnerConfig.java | 6 +-
.../MultipleKubernetesTaskRunnerFactory.java | 6 +-
.../k8s/overlord/KubernetesPeonLifecycleTest.java | 74 ++++++++++++
.../KubernetesTaskRunnerStaticConfigTest.java | 14 +++
.../MultipleKubernetesTaskRunnerConfigTest.java | 12 ++
.../druid/indexing/overlord/ForkingTaskRunner.java | 2 +
.../overlord/SingleTaskBackgroundRunner.java | 2 +-
.../indexing/overlord/ThreadingTaskRunner.java | 2 +-
.../indexing/overlord/ForkingTaskRunnerTest.java | 66 +++++++++++
.../overlord/SingleTaskBackgroundRunnerTest.java | 13 +++
.../indexing/overlord/ThreadingTaskRunnerTest.java | 50 +++++++++
.../java/org/apache/druid/rpc/ServiceLocation.java | 2 +-
.../java/org/apache/druid/server/DruidNode.java | 71 ++++++++++--
.../druid/server/http/BaseDynamicConfigSyncer.java | 7 +-
.../org/apache/druid/rpc/ServiceLocationTest.java | 18 +++
.../org/apache/druid/server/DruidNodeTest.java | 125 +++++++++++++++++++++
.../http/CoordinatorDynamicConfigSyncerTest.java | 42 ++++++-
.../router/TieredBrokerHostSelectorTest.java | 7 +-
28 files changed, 592 insertions(+), 35 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 4eff8506d41..f46f919e454 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -167,6 +167,22 @@ Druid interacts with ZooKeeper through a set of standard
path configurations. We
Although not recommended but both HTTP and HTTPS connectors can be enabled at
a time and respective ports are configurable using `druid.plaintextPort`
and `druid.tlsPort` properties on each service. Please see `Configuration`
section of individual services to check the valid and default values for these
ports.
+#### Advertised plaintext port
+
+By default, a service advertises the same plaintext port it listens on. Set
`druid.advertisedPlaintextPort` when other services must reach it on a
different port, for example when a sidecar proxy such as Envoy terminates mTLS
on one port and forwards traffic to Jetty on `druid.plaintextPort`.
+
+- `druid.plaintextPort` remains the port that Jetty binds to and listens on.
+- `druid.advertisedPlaintextPort` is the port that the service publishes
through service discovery and that other services use to reach it over
plaintext. This covers internal RPC between services, Router to Broker
proxying, dynamic configuration sync, lookup and security cache notifications,
catalog sync, `sys.server_properties`, task locations reported by Indexers, and
Consul registration.
+- If `druid.advertisedPlaintextPort` is unset or non-positive, it falls back
to `druid.plaintextPort`.
+- It only applies when `druid.enablePlaintextPort` is `true`. When
`druid.enableTlsPort` is `true`, services still prefer `druid.tlsPort` to reach
each other.
+- The `plaintext_port` column of `sys.servers` and the `plaintextPort` field
of the `/druid/coordinator/v1/cluster` API continue to report
`druid.plaintextPort`.
+
+You can't use a port in `druid.host` for this purpose, because a port in
`druid.host` must match `druid.plaintextPort`.
+
+Peons forked by a Middle Manager don't support
`druid.advertisedPlaintextPort`. Each Peon listens on its own port from
`druid.indexer.runner.startPort`, `druid.indexer.runner.endPort`, or
`druid.indexer.runner.ports`, so a single advertised port can't identify an
individual Peon. The Middle Manager always starts Peons with
`druid.advertisedPlaintextPort` unset, so each Peon advertises its own port.
This applies even if the property is inherited from the Middle Manager or set
through `dru [...]
+
+Task pods launched by the [Kubernetes task
runner](../development/extensions-core/k8s-jobs.md) aren't reached through
service discovery. The Overlord reports each task's location from the pod IP
and a fixed port, so `druid.advertisedPlaintextPort` has no effect on them. To
reach task pods through a sidecar proxy, set
[`druid.indexer.runner.advertisedPlaintextPort`](../development/extensions-core/k8s-jobs.md#properties)
on the Overlord instead.
+
#### Jetty server TLS configuration
Druid uses Jetty as an embedded web server. To learn more about TLS/SSL,
certificates, and related concepts in Jetty, including explanations of the
configuration settings below, see "Configuring SSL/TLS KeyStores" in the [Jetty
Operations Guide](https://www.eclipse.org/jetty/documentation.php).
@@ -718,6 +734,7 @@ These Coordinator static configurations can be defined in
the `coordinator/runti
|`druid.host`|The host for the current service. This is used to advertise the
current service location as reachable from another service and should generally
be specified such that `http://${druid.host}/` could actually talk to this
service.|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the service's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8081|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative integer.|8281|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services.|`druid/coordinator`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -967,6 +984,7 @@ These Overlord static configurations can be defined in the
`overlord/runtime.pro
|`druid.host`|The host for the current service. This is used to advertise the
current service location as reachable from another service and should generally
be specified such that `http://${druid.host}/` could actually talk to this
service.|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the service's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`.|8090|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|8290|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services.|`druid/overlord`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -1319,6 +1337,7 @@ These Middle Manager and Peon configurations can be
defined in the `middleManage
|`druid.host`|The host for the current service. This is used to advertise the
current service location as reachable from another service and should generally
be specified such that `http://${druid.host}/` could actually talk to this
service|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the service's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8091|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. Ignored for Peons forked by a Middle Manager, which
always advertise their own port. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|8291|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services|`druid/middlemanager`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -1446,6 +1465,7 @@ For most types of tasks, `SegmentWriteOutMediumFactory`
can be configured per-ta
|`druid.host`|The host for the current process. This is used to advertise the
current processes location as reachable from another process and should
generally be specified such that `http://${druid.host}/` could actually talk to
this process|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the process's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8091|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|8283|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services|`druid/indexer`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -1544,6 +1564,7 @@ These Historical configurations can be defined in the
`historical/runtime.proper
|`druid.host`|The host for the current service. This is used to advertise the
current service location as reachable from another service and should generally
be specified such that `http://${druid.host}/` could actually talk to this
service|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the service's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8083|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|8283|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services|`druid/historical`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -1665,6 +1686,7 @@ These Broker configurations can be defined in the
`broker/runtime.properties` fi
|`druid.host`|The host for the current process. This is used to advertise the
current processes location as reachable from another process and should
generally be specified such that `http://${druid.host}/` could actually talk to
this process|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the process's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8082|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|8282|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services|`druid/broker`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
@@ -2333,6 +2355,7 @@ Supported query contexts:
|`druid.host`|The host for the current process. This is used to advertise the
current processes location as reachable from another process and should
generally be specified such that `http://${druid.host}/` could actually talk to
this process|`InetAddress.getLocalHost().getCanonicalHostName()`|
|`druid.bindOnHost`|Indicating whether the process's internal jetty server
bind on `druid.host`. Default is false, which means binding to all
interfaces.|false|
|`druid.plaintextPort`|This is the port to actually listen on; unless port
mapping is used, this will be the same port as is on `druid.host`|8888|
+|`druid.advertisedPlaintextPort`|Plaintext port that other services use to
reach this service, if different from `druid.plaintextPort`, for example when a
sidecar proxy listens in front of Jetty. Unset or non-positive values fall back
to `druid.plaintextPort`. See [Advertised plaintext
port](#advertised-plaintext-port).|`druid.plaintextPort`|
|`druid.tlsPort`|TLS port for HTTPS connector, if
[druid.enableTlsPort](../operations/tls-support.md) is set then this config
will be used. If `druid.host` contains port then that port will be ignored.
This should be a non-negative Integer.|9088|
|`druid.service`|The name of the service. This is used as a dimension when
emitting metrics and alerts to differentiate between the various
services|`druid/router`|
|`druid.labels`|Optional JSON object of key-value pairs that define custom
labels for the server. These labels are displayed in the web console under the
"Services" tab. Example: `druid.labels={"location":"Airtrunk"}` or
`druid.labels.location=Airtrunk`|`null`|
diff --git a/docs/development/extensions-core/k8s-jobs.md
b/docs/development/extensions-core/k8s-jobs.md
index 242d17a9924..2906ea75fa3 100644
--- a/docs/development/extensions-core/k8s-jobs.md
+++ b/docs/development/extensions-core/k8s-jobs.md
@@ -1033,6 +1033,7 @@ the Overlord pod from
`druid.indexer.runner.overlordNamespace` or `druid.indexer
| `druid.indexer.runner.logSaveTimeout` | `Duration` | The peon executing the
ingestion task makes a best effort to persist the pod logs from `k8s` to
persistent task log storage. The timeout ensures that `k8s` connection issues
do not cause the pod to hang indefinitely thereby blocking Overlord operations.
If the timeout occurs before the logs are saved, those logs will not be
available in Druid. | `PT300S` | NO |
| `druid.indexer.runner.useK8sSharedInformers` | `boolean` | Whether to use
shared informers to watch for pod/job changes. This is more efficient on the
Kubernetes API server, but may use more memory in the Overlord. | `false` | No |
| `druid.indexer.runner.k8sSharedInformerResyncPeriod` | `Duration` | When
using shared informers, controls how frequently the informers resync with the
Kubernetes API server. This prevents change events from being missed, keeping
the informer cache clean and accurate. | `PT300S` | No |
+| `druid.indexer.runner.advertisedPlaintextPort` | `Integer` | Plaintext port
reported in task locations, if other services must reach task pods on a port
other than the one the peon listens on, for example when a sidecar proxy in the
task pod listens in front of the peon. Task pods aren't reached through service
discovery, so `druid.advertisedPlaintextPort` doesn't apply to them. Unset or
non-positive values use the peon's own port. | `8100` | No |
| `druid.indexer.runner.clusterSelector.type` | `String` (`roundrobin`,
`random`, `leastTask`) | Only applicable when `druid.indexer.runner.type` is
`multik8s`. Strategy used to select which enabled Kubernetes cluster receives
the next task. | `roundrobin` | No |
| `druid.indexer.runner.clusters[N].name` | `String` | Only applicable when
`druid.indexer.runner.type` is `multik8s`. Human-readable name for cluster `N`.
When set, Druid adds this value to task context tags as `k8s_cluster`. | `null`
| No |
| `druid.indexer.runner.clusters[N].taskNamespace` | `String` | Only
applicable when `druid.indexer.runner.type` is `multik8s`. Kubernetes namespace
where task pods and jobs run for cluster `N`. | - | Yes |
diff --git a/docs/querying/sql-metadata-tables.md
b/docs/querying/sql-metadata-tables.md
index 055305d81e5..81d3f6bdb26 100644
--- a/docs/querying/sql-metadata-tables.md
+++ b/docs/querying/sql-metadata-tables.md
@@ -228,7 +228,7 @@ Servers table lists all discovered servers in the cluster.
|------|-----|-----|
|server|VARCHAR|Server name in the form host:port|
|host|VARCHAR|Hostname of the server|
-|plaintext_port|BIGINT|Unsecured port of the server, or -1 if plaintext
traffic is disabled|
+|plaintext_port|BIGINT|Unsecured port the server listens on
(`druid.plaintextPort`), or -1 if plaintext traffic is disabled. The port in
`server` may differ if
[`druid.advertisedPlaintextPort`](../configuration/index.md#advertised-plaintext-port)
is set.|
|tls_port|BIGINT|TLS port of the server, or -1 if TLS is disabled|
|server_type|VARCHAR|Type of Druid service. Possible values include:
COORDINATOR, OVERLORD, BROKER, ROUTER, HISTORICAL, MIDDLE_MANAGER or PEON.|
|tier|VARCHAR|Distribution tier see
[druid.server.tier](../configuration/index.md#historical-general-configuration).
Only valid for HISTORICAL type, for other types it's null|
diff --git
a/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulServiceIdsTest.java
b/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulServiceIdsTest.java
index aa41963f994..92d7c461bcc 100644
---
a/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulServiceIdsTest.java
+++
b/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulServiceIdsTest.java
@@ -59,5 +59,11 @@ public class ConsulServiceIdsTest
String kvKey = ConsulServiceIds.nodeKvKey(config, serviceId);
Assertions.assertEquals("druid/nodes/druid-peon-host-8080", kvKey);
+
+ DruidNode advertisedNode = new DruidNode("service", "host", false, 8080,
null, null, true, false, null, 9080);
+ Assertions.assertEquals(
+ "druid-peon-host-9080",
+ ConsulServiceIds.serviceId(config, new
DiscoveryDruidNode(advertisedNode, role, null))
+ );
}
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
index d15fa396323..70af7cbda27 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java
@@ -100,6 +100,7 @@ public class KubernetesPeonLifecycle
private final TaskStateListener stateListener;
private final SettableFuture<Boolean> taskStartedSuccessfullyFuture;
private final long logSaveTimeoutMs;
+ private final int plaintextPort;
@MonotonicNonNull
private LogWatch logWatch;
@@ -116,6 +117,23 @@ public class KubernetesPeonLifecycle
long logSaveTimeoutMs
)
{
+ this(task, taskId, kubernetesClient, taskLogs, mapper, stateListener,
logSaveTimeoutMs, null);
+ }
+
+ protected KubernetesPeonLifecycle(
+ Task task,
+ K8sTaskId taskId,
+ KubernetesPeonClient kubernetesClient,
+ TaskLogs taskLogs,
+ ObjectMapper mapper,
+ TaskStateListener stateListener,
+ long logSaveTimeoutMs,
+ @Nullable Integer advertisedPlaintextPort
+ )
+ {
+ this.plaintextPort = advertisedPlaintextPort != null &&
advertisedPlaintextPort > 0
+ ? advertisedPlaintextPort
+ : DruidK8sConstants.PORT;
this.task = task;
this.taskId = taskId;
this.kubernetesClient = kubernetesClient;
@@ -300,7 +318,7 @@ public class KubernetesPeonLifecycle
}
taskLocationRef.set(TaskLocation.create(
podStatus.getPodIP(),
- DruidK8sConstants.PORT,
+ plaintextPort,
DruidK8sConstants.TLS_PORT,
Boolean.parseBoolean(pod.getMetadata().getAnnotations().getOrDefault(DruidK8sConstants.TLS_ENABLED,
"false")),
pod.getMetadata() != null ? pod.getMetadata().getName() : ""
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleFactory.java
index 8bd7db2ebf3..30a78961f92 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleFactory.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleFactory.java
@@ -25,24 +25,30 @@ import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.common.KubernetesPeonClient;
import org.apache.druid.tasklogs.TaskLogs;
+import javax.annotation.Nullable;
+
public class KubernetesPeonLifecycleFactory implements PeonLifecycleFactory
{
private final KubernetesPeonClient client;
private final TaskLogs taskLogs;
private final ObjectMapper mapper;
private final long logSaveTimeoutMs;
+ @Nullable
+ private final Integer advertisedPlaintextPort;
public KubernetesPeonLifecycleFactory(
KubernetesPeonClient client,
TaskLogs taskLogs,
ObjectMapper mapper,
- long logSaveTimeoutMs
+ long logSaveTimeoutMs,
+ @Nullable Integer advertisedPlaintextPort
)
{
this.client = client;
this.taskLogs = taskLogs;
this.mapper = mapper;
this.logSaveTimeoutMs = logSaveTimeoutMs;
+ this.advertisedPlaintextPort = advertisedPlaintextPort;
}
@Override
@@ -55,7 +61,8 @@ public class KubernetesPeonLifecycleFactory implements
PeonLifecycleFactory
taskLogs,
mapper,
stateListener,
- logSaveTimeoutMs
+ logSaveTimeoutMs,
+ advertisedPlaintextPort
);
}
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java
index d164db06298..84c516d8daf 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java
@@ -23,6 +23,7 @@ import jakarta.validation.constraints.Max;
import jakarta.validation.constraints.Min;
import org.joda.time.Period;
+import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
@@ -96,6 +97,12 @@ public interface KubernetesTaskRunnerConfig
*/
boolean isAllowTaskPodTemplateSelection();
+ /**
+ * Plaintext port reported in task pod {@code TaskLocation}s, e.g. a sidecar
proxy port; null uses {@code DruidK8sConstants.PORT}.
+ */
+ @Nullable
+ Integer getAdvertisedPlaintextPort();
+
static Builder builder()
{
return new Builder();
@@ -127,6 +134,7 @@ public interface KubernetesTaskRunnerConfig
private boolean useK8sSharedInformers;
private Period k8sSharedInformerResyncPeriod;
private boolean allowTaskPodTemplateSelection;
+ private Integer advertisedPlaintextPort;
public Builder()
{
@@ -277,6 +285,12 @@ public interface KubernetesTaskRunnerConfig
return this;
}
+ public Builder withAdvertisedPlaintextPort(Integer advertisedPlaintextPort)
+ {
+ this.advertisedPlaintextPort = advertisedPlaintextPort;
+ return this;
+ }
+
public KubernetesTaskRunnerStaticConfig build()
{
return new KubernetesTaskRunnerStaticConfig(
@@ -303,7 +317,8 @@ public interface KubernetesTaskRunnerConfig
this.taskJoinTimeout,
this.useK8sSharedInformers,
this.k8sSharedInformerResyncPeriod,
- this.allowTaskPodTemplateSelection
+ this.allowTaskPodTemplateSelection,
+ this.advertisedPlaintextPort
);
}
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerEffectiveConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerEffectiveConfig.java
index 227b78d7e84..068670286d7 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerEffectiveConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerEffectiveConfig.java
@@ -24,6 +24,7 @@ import
org.apache.druid.k8s.overlord.execution.KubernetesTaskRunnerDynamicConfig
import org.apache.druid.k8s.overlord.execution.PodTemplateSelectStrategy;
import org.joda.time.Period;
+import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
@@ -195,6 +196,13 @@ public class KubernetesTaskRunnerEffectiveConfig
implements KubernetesTaskRunner
return staticConfig.isAllowTaskPodTemplateSelection();
}
+ @Override
+ @Nullable
+ public Integer getAdvertisedPlaintextPort()
+ {
+ return staticConfig.getAdvertisedPlaintextPort();
+ }
+
public PodTemplateSelectStrategy getPodTemplateSelectStrategy()
{
if (dynamicConfigSupplier == null || dynamicConfigSupplier.get() == null
|| dynamicConfigSupplier.get().getPodTemplateSelectStrategy() == null) {
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactory.java
index 325f16eed72..0a753f6b213 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactory.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactory.java
@@ -111,7 +111,8 @@ public class KubernetesTaskRunnerFactory implements
TaskRunnerFactory<Kubernetes
peonClient,
taskLogs,
smileMapper,
-
kubernetesTaskRunnerConfig.getLogSaveTimeout().toStandardDuration().getMillis()
+
kubernetesTaskRunnerConfig.getLogSaveTimeout().toStandardDuration().getMillis(),
+ kubernetesTaskRunnerConfig.getAdvertisedPlaintextPort()
),
emitter,
configManager
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfig.java
index b8ab038ca94..97cfe3f969c 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfig.java
@@ -29,6 +29,7 @@ import org.apache.commons.lang3.ObjectUtils;
import org.joda.time.Period;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
@@ -156,6 +157,11 @@ public class KubernetesTaskRunnerStaticConfig implements
KubernetesTaskRunnerCon
// Allow per-task pod template selection via the task context.
private boolean allowTaskPodTemplateSelection = false;
+ @JsonProperty
+ @Max(0xffff)
+ // task pods are reached by pod IP, not service discovery, so
druid.advertisedPlaintextPort can't apply to them
+ private Integer advertisedPlaintextPort = null;
+
public KubernetesTaskRunnerStaticConfig()
{
}
@@ -184,7 +190,8 @@ public class KubernetesTaskRunnerStaticConfig implements
KubernetesTaskRunnerCon
Period taskJoinTimeout,
boolean useK8sSharedInformers,
Period k8sSharedInformerResyncPeriod,
- boolean allowTaskPodTemplateSelection
+ boolean allowTaskPodTemplateSelection,
+ @Nullable Integer advertisedPlaintextPort
)
{
this.namespace = namespace;
@@ -274,6 +281,7 @@ public class KubernetesTaskRunnerStaticConfig implements
KubernetesTaskRunnerCon
allowTaskPodTemplateSelection,
this.allowTaskPodTemplateSelection
);
+ this.advertisedPlaintextPort = advertisedPlaintextPort;
}
@Override
@@ -421,4 +429,11 @@ public class KubernetesTaskRunnerStaticConfig implements
KubernetesTaskRunnerCon
{
return allowTaskPodTemplateSelection;
}
+
+ @Override
+ @Nullable
+ public Integer getAdvertisedPlaintextPort()
+ {
+ return advertisedPlaintextPort;
+ }
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfig.java
index e6da1cb39fe..38d8c0a6624 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfig.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfig.java
@@ -148,7 +148,8 @@ public class MultipleKubernetesTaskRunnerConfig extends
KubernetesTaskRunnerStat
@JsonProperty("taskJoinTimeout") Period taskJoinTimeout,
@JsonProperty("useK8sSharedInformers") Boolean useK8sSharedInformers,
@JsonProperty("k8sSharedInformerResyncPeriod") Period
k8sSharedInformerResyncPeriod,
- @JsonProperty("allowTaskPodTemplateSelection") Boolean
allowTaskPodTemplateSelection
+ @JsonProperty("allowTaskPodTemplateSelection") Boolean
allowTaskPodTemplateSelection,
+ @JsonProperty("advertisedPlaintextPort") Integer advertisedPlaintextPort
)
{
super(
@@ -175,7 +176,8 @@ public class MultipleKubernetesTaskRunnerConfig extends
KubernetesTaskRunnerStat
taskJoinTimeout,
useK8sSharedInformers != null && useK8sSharedInformers,
k8sSharedInformerResyncPeriod,
- allowTaskPodTemplateSelection != null && allowTaskPodTemplateSelection
+ allowTaskPodTemplateSelection != null && allowTaskPodTemplateSelection,
+ advertisedPlaintextPort
);
Preconditions.checkNotNull(clusters);
Preconditions.checkState(
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactory.java
index 46a9008a986..e90c3bd84ae 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactory.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactory.java
@@ -179,7 +179,8 @@ public class MultipleKubernetesTaskRunnerFactory implements
TaskRunnerFactory<Ta
peonClient,
taskLogs,
smileMapper,
-
effectiveConfig.getLogSaveTimeout().toStandardDuration().getMillis()
+
effectiveConfig.getLogSaveTimeout().toStandardDuration().getMillis(),
+ effectiveConfig.getAdvertisedPlaintextPort()
),
emitter,
sharedExecutor,
@@ -467,7 +468,8 @@ public class MultipleKubernetesTaskRunnerFactory implements
TaskRunnerFactory<Ta
this.runnerConfig.getTaskJoinTimeout(),
this.runnerConfig.isUseK8sSharedInformers(),
this.runnerConfig.getK8sSharedInformerResyncPeriod(),
- this.runnerConfig.isAllowTaskPodTemplateSelection()
+ this.runnerConfig.isAllowTaskPodTemplateSelection(),
+ this.runnerConfig.getAdvertisedPlaintextPort()
);
}
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
index 7df86556bb4..ba8806928c8 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycleTest.java
@@ -31,6 +31,7 @@ import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.TestUtils;
import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.k8s.overlord.common.DruidK8sConstants;
import org.apache.druid.k8s.overlord.common.JobResponse;
import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.common.K8sTestUtils;
@@ -956,6 +957,79 @@ public class KubernetesPeonLifecycleTest extends
EasyMockSupport
verifyAll();
}
+ @Test
+ public void
test_getTaskLocation_withAdvertisedPlaintextPort_returnsAdvertisedPort()
+ throws NoSuchFieldException, IllegalAccessException
+ {
+ KubernetesPeonLifecycle peonLifecycle = new KubernetesPeonLifecycle(
+ task,
+ k8sTaskId,
+ kubernetesClient,
+ taskLogs,
+ mapper,
+ stateListener,
+ LOG_SAVE_TIMEOUT.toStandardDuration().getMillis(),
+ 9443
+ );
+ setPeonLifecycleState(peonLifecycle,
KubernetesPeonLifecycle.State.RUNNING);
+
+ Pod pod = new PodBuilder()
+ .withNewMetadata()
+ .withName(ID)
+ .endMetadata()
+ .withNewStatus()
+ .withPodIP("ip")
+ .endStatus()
+ .build();
+
+
EasyMock.expect(kubernetesClient.getPeonPod(k8sTaskId.getK8sJobName())).andReturn(Optional.of(pod));
+
+ replayAll();
+
+ TaskLocation location = peonLifecycle.getTaskLocation();
+
+ Assertions.assertEquals("ip", location.getHost());
+ Assertions.assertEquals(9443, location.getPort());
+ Assertions.assertEquals(-1, location.getTlsPort());
+ Assertions.assertEquals(ID, location.getK8sPodName());
+
+ verifyAll();
+ }
+
+ @Test
+ public void
test_getTaskLocation_withNonPositiveAdvertisedPlaintextPort_returnsDefaultPort()
+ throws NoSuchFieldException, IllegalAccessException
+ {
+ KubernetesPeonLifecycle peonLifecycle = new KubernetesPeonLifecycle(
+ task,
+ k8sTaskId,
+ kubernetesClient,
+ taskLogs,
+ mapper,
+ stateListener,
+ LOG_SAVE_TIMEOUT.toStandardDuration().getMillis(),
+ -1
+ );
+ setPeonLifecycleState(peonLifecycle,
KubernetesPeonLifecycle.State.RUNNING);
+
+ Pod pod = new PodBuilder()
+ .withNewMetadata()
+ .withName(ID)
+ .endMetadata()
+ .withNewStatus()
+ .withPodIP("ip")
+ .endStatus()
+ .build();
+
+
EasyMock.expect(kubernetesClient.getPeonPod(k8sTaskId.getK8sJobName())).andReturn(Optional.of(pod));
+
+ replayAll();
+
+ Assertions.assertEquals(DruidK8sConstants.PORT,
peonLifecycle.getTaskLocation().getPort());
+
+ verifyAll();
+ }
+
@Test
public void test_getTaskLocation_saveTaskLocation()
throws NoSuchFieldException, IllegalAccessException
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfigTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfigTest.java
index 7c0338d1db6..9754da7653c 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfigTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerStaticConfigTest.java
@@ -80,6 +80,18 @@ public class KubernetesTaskRunnerStaticConfigTest
Assertions.assertEquals(ImmutableMap.of(), config.getLabels());
Assertions.assertEquals(ImmutableMap.of(), config.getAnnotations());
Assertions.assertEquals(Integer.valueOf(Integer.MAX_VALUE),
config.getCapacity());
+ Assertions.assertNull(config.getAdvertisedPlaintextPort());
+ }
+
+ @Test
+ public void test_deserializable_withAdvertisedPlaintextPort() throws
IOException
+ {
+ KubernetesTaskRunnerStaticConfig config = new
DefaultObjectMapper().readValue(
+ "{\"namespace\": \"namespace\", \"advertisedPlaintextPort\": 9443}",
+ KubernetesTaskRunnerStaticConfig.class
+ );
+
+ Assertions.assertEquals(Integer.valueOf(9443),
config.getAdvertisedPlaintextPort());
}
@Test
@@ -102,6 +114,7 @@ public class KubernetesTaskRunnerStaticConfigTest
.withLabels(ImmutableMap.of("key", "value"))
.withAnnotations(ImmutableMap.of("key", "value"))
.withCapacity(1)
+ .withAdvertisedPlaintextPort(9443)
.build();
Assertions.assertEquals("namespace", config.getNamespace());
@@ -119,5 +132,6 @@ public class KubernetesTaskRunnerStaticConfigTest
Assertions.assertEquals(ImmutableMap.of("key", "value"),
config.getLabels());
Assertions.assertEquals(ImmutableMap.of("key", "value"),
config.getAnnotations());
Assertions.assertEquals(Integer.valueOf(1), config.getCapacity());
+ Assertions.assertEquals(Integer.valueOf(9443),
config.getAdvertisedPlaintextPort());
}
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfigTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfigTest.java
index 3dcfeb610d8..d7f0317a239 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfigTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerConfigTest.java
@@ -220,6 +220,18 @@ public class MultipleKubernetesTaskRunnerConfigTest
Assertions.assertTrue(config.isAllowTaskPodTemplateSelection());
}
+ @Test
+ public void test_fromProperties_withAdvertisedPlaintextPort()
+ {
+ final Properties props = new Properties();
+ props.setProperty("druid.indexer.runner.clusters[0].taskNamespace",
"namespace1");
+ props.setProperty("druid.indexer.runner.advertisedPlaintextPort", "9443");
+
+ final MultipleKubernetesTaskRunnerConfig config =
MultipleKubernetesTaskRunnerConfig.fromProperties(props);
+
+ Assertions.assertEquals(Integer.valueOf(9443),
config.getAdvertisedPlaintextPort());
+ }
+
@Test
public void
test_fromProperties_withoutSelectionStrategy_defaultsToRoundRobin()
{
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
index 606f5152063..01b480a0516 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
@@ -355,6 +355,8 @@ public class ForkingTaskRunner
command.addSystemProperty("druid.host", childHost);
command.addSystemProperty("druid.plaintextPort",
childPort);
+ // Forked peons listen on their own port; never let
them advertise an inherited one.
+
command.addSystemProperty("druid.advertisedPlaintextPort", -1);
command.addSystemProperty("druid.tlsPort",
tlsChildPort);
// Let tasks know where they are running on.
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
index 361c431c762..8e0ed1c33cb 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
@@ -98,7 +98,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
this.toolboxFactory = Preconditions.checkNotNull(toolboxFactory,
"toolboxFactory");
this.taskConfig = taskConfig;
this.emitter = Preconditions.checkNotNull(emitter, "emitter");
- this.location = TaskLocation.create(node.getHost(),
node.getPlaintextPort(), node.getTlsPort());
+ this.location = TaskLocation.create(node.getHost(),
node.getAdvertisedPlaintextPort(), node.getTlsPort());
this.serverConfig = serverConfig;
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
index 999e36a3dd3..b057e0409ef 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
@@ -182,7 +182,7 @@ public class ThreadingTaskRunner
final TaskLocation taskLocation =
TaskLocation.create(
node.getHost(),
- node.getPlaintextPort(),
+ node.getAdvertisedPlaintextPort(),
node.getTlsPort()
);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java
index 51ffe8ac337..26b5a50ff97 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java
@@ -67,10 +67,12 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Paths;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -528,6 +530,70 @@ public class ForkingTaskRunnerTest
Assertions.assertTrue(xmxJavaOptsIndex.get() >= 0);
}
+ @Test
+ public void testChildNeverInheritsAdvertisedPlaintextPort() throws
ExecutionException, InterruptedException
+ {
+ final Properties props = new Properties();
+ props.setProperty("druid.advertisedPlaintextPort", "9443");
+
props.setProperty("druid.indexer.fork.property.druid.advertisedPlaintextPort",
"9444");
+ final Task task = new NoopTask(
+ null,
+ null,
+ null,
+ 1L,
+ 0L,
+ Map.of("druid.indexer.fork.property.druid.advertisedPlaintextPort",
9445)
+ );
+ final AtomicReference<List<String>> childCommand = new AtomicReference<>();
+ final TaskConfig taskConfig = makeDefaultTaskConfigBuilder().build();
+ final WorkerConfig workerConfig = new WorkerConfig();
+ final ForkingTaskRunner forkingTaskRunner = new ForkingTaskRunner(
+ new ForkingTaskRunnerConfig(),
+ taskConfig,
+ workerConfig,
+ props,
+ new NoopTaskLogs(),
+ OBJECT_MAPPER,
+ new DruidNode("middleManager", "host", false, 8091, null, null, true,
false, null, 9443),
+ new StartupLoggingConfig(),
+ TaskStorageDirTracker.fromConfigs(workerConfig, taskConfig)
+ )
+ {
+ @Override
+ ProcessHolder runTaskProcess(List<String> command, File logFile,
TaskLocation taskLocation)
+ {
+ childCommand.set(command);
+ return makeTestProcessHolder(logFile, taskLocation);
+ }
+
+ @Override
+ int waitForTaskProcessToComplete(Task task, ProcessHolder processHolder,
File logFile, File reportsFile)
+ {
+ return 1;
+ }
+ };
+
+ forkingTaskRunner.setNumProcessorsPerTask();
+ forkingTaskRunner.run(task).get();
+
+ final List<String> advertisedArgs = childCommand.get()
+ .stream()
+ .filter(arg ->
arg.startsWith("-Ddruid.advertisedPlaintextPort="))
+
.collect(Collectors.toList());
+ Assertions.assertTrue(
+ advertisedArgs.containsAll(
+ List.of(
+ "-Ddruid.advertisedPlaintextPort=9443",
+ "-Ddruid.advertisedPlaintextPort=9444",
+ "-Ddruid.advertisedPlaintextPort=9445"
+ )
+ ),
+ advertisedArgs.toString()
+ );
+ // The JVM keeps the last -D value, so the forced -1 must come after every
inherited or overridden value.
+ Assertions.assertEquals("-Ddruid.advertisedPlaintextPort=-1",
advertisedArgs.get(advertisedArgs.size() - 1));
+ }
+
@Test
public void testInvalidTaskContextJavaOptsArray() throws
JsonProcessingException
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
index 0a73e82ad31..f44f5b194e0 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
@@ -159,6 +159,19 @@ public class SingleTaskBackgroundRunnerTest
runner.stop();
}
+ @Test
+ public void testTaskLocationUsesAdvertisedPlaintextPort()
+ {
+ final SingleTaskBackgroundRunner advertisedRunner = new
SingleTaskBackgroundRunner(
+ EasyMock.createMock(TaskToolboxFactory.class),
+ new TaskConfigBuilder().build(),
+ new NoopServiceEmitter(),
+ new DruidNode("testServer", "testHost", false, 1000, null, null, true,
false, null, 2000),
+ new ServerConfig()
+ );
+ Assertions.assertEquals(TaskLocation.create("testHost", 2000, -1),
advertisedRunner.getTaskLocation("any"));
+ }
+
@Test
public void testRun() throws ExecutionException, InterruptedException
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ThreadingTaskRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ThreadingTaskRunnerTest.java
index 968051d9339..c9fc2c40d99 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ThreadingTaskRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/ThreadingTaskRunnerTest.java
@@ -21,6 +21,7 @@ package org.apache.druid.indexing.overlord;
import com.google.common.base.Optional;
import org.apache.commons.io.IOUtils;
+import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.MultipleFileTaskReportFileWriter;
@@ -34,6 +35,7 @@ import
org.apache.druid.indexing.common.task.TestAppenderatorsManager;
import org.apache.druid.indexing.worker.config.WorkerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.server.DruidNode;
import org.apache.druid.tasklogs.NoopTaskLogs;
@@ -43,7 +45,9 @@ import org.junit.jupiter.api.Test;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -141,6 +145,52 @@ public class ThreadingTaskRunnerTest
);
}
+ @Test
+ public void testTaskLocationUsesAdvertisedPlaintextPort() throws Exception
+ {
+ final TaskConfig taskConfig =
ForkingTaskRunnerTest.makeDefaultTaskConfigBuilder().build();
+ final WorkerConfig workerConfig = new WorkerConfig();
+ runner = new ThreadingTaskRunner(
+ mockTaskToolboxFactory(),
+ taskConfig,
+ workerConfig,
+ new NoopTaskLogs(),
+ new DefaultObjectMapper(),
+ new TestAppenderatorsManager(),
+ new MultipleFileTaskReportFileWriter(),
+ new DruidNode("druid/indexer", "host", false, 8091, null, null, true,
false, null, 9091),
+ TaskStorageDirTracker.fromConfigs(workerConfig, taskConfig)
+ );
+
+ final List<TaskLocation> locations = new CopyOnWriteArrayList<>();
+ runner.registerListener(
+ new TaskRunnerListener()
+ {
+ @Override
+ public String getListenerId()
+ {
+ return "testTaskLocationUsesAdvertisedPlaintextPort";
+ }
+
+ @Override
+ public void locationChanged(String taskId, TaskLocation newLocation)
+ {
+ locations.add(newLocation);
+ }
+
+ @Override
+ public void statusChanged(String taskId, TaskStatus status)
+ {
+ }
+ },
+ Execs.directExecutor()
+ );
+
+ final TaskStatus status = runner.run(new NoopTask(null, null, null, 1L,
0L, Map.of())).get();
+ Assertions.assertEquals(TaskState.SUCCESS, status.getStatusCode());
+ Assertions.assertEquals(List.of(TaskLocation.create("host", 9091, -1)),
locations);
+ }
+
private static TaskToolboxFactory mockTaskToolboxFactory()
{
return new TestTaskToolboxFactory(new TestTaskToolboxFactory.Builder());
diff --git a/server/src/main/java/org/apache/druid/rpc/ServiceLocation.java
b/server/src/main/java/org/apache/druid/rpc/ServiceLocation.java
index ab588a0736e..971e6554d02 100644
--- a/server/src/main/java/org/apache/druid/rpc/ServiceLocation.java
+++ b/server/src/main/java/org/apache/druid/rpc/ServiceLocation.java
@@ -73,7 +73,7 @@ public class ServiceLocation
*/
public static ServiceLocation fromDruidNode(final DruidNode druidNode)
{
- return new ServiceLocation(druidNode.getHost(),
druidNode.getPlaintextPort(), druidNode.getTlsPort(), "");
+ return new ServiceLocation(druidNode.getHost(),
druidNode.getAdvertisedPlaintextPort(), druidNode.getTlsPort(), "");
}
/**
diff --git a/server/src/main/java/org/apache/druid/server/DruidNode.java
b/server/src/main/java/org/apache/druid/server/DruidNode.java
index f76edcfffcc..e3f7f0c464d 100644
--- a/server/src/main/java/org/apache/druid/server/DruidNode.java
+++ b/server/src/main/java/org/apache/druid/server/DruidNode.java
@@ -100,6 +100,10 @@ public class DruidNode
@JsonProperty
private Map<String, String> labels;
+ @JsonProperty
+ @Max(0xffff)
+ private int advertisedPlaintextPort = -1;
+
public DruidNode(
String serviceName,
String host,
@@ -110,7 +114,36 @@ public class DruidNode
boolean enableTlsPort
)
{
- this(serviceName, host, bindOnHost, plaintextPort, null, tlsPort,
enablePlaintextPort, enableTlsPort, null);
+ this(serviceName, host, bindOnHost, plaintextPort, null, tlsPort,
enablePlaintextPort, enableTlsPort, null, null);
+ }
+
+ public DruidNode(
+ String serviceName,
+ String host,
+ boolean bindOnHost,
+ Integer plaintextPort,
+ Integer port,
+ Integer tlsPort,
+ Boolean enablePlaintextPort,
+ boolean enableTlsPort
+ )
+ {
+ this(serviceName, host, bindOnHost, plaintextPort, port, tlsPort,
enablePlaintextPort, enableTlsPort, null, null);
+ }
+
+ public DruidNode(
+ String serviceName,
+ String host,
+ boolean bindOnHost,
+ Integer plaintextPort,
+ Integer port,
+ Integer tlsPort,
+ Boolean enablePlaintextPort,
+ boolean enableTlsPort,
+ Map<String, String> labels
+ )
+ {
+ this(serviceName, host, bindOnHost, plaintextPort, port, tlsPort,
enablePlaintextPort, enableTlsPort, labels, null);
}
/**
@@ -139,7 +172,8 @@ public class DruidNode
@JacksonInject(useInput = OptBoolean.TRUE) @Named("tlsServicePort")
@JsonProperty("tlsPort") Integer tlsPort,
@JsonProperty("enablePlaintextPort") Boolean enablePlaintextPort,
@JsonProperty("enableTlsPort") boolean enableTlsPort,
- @JsonProperty("labels") @Nullable Map<String, String> labels
+ @JsonProperty("labels") @Nullable Map<String, String> labels,
+ @JsonProperty("advertisedPlaintextPort") Integer advertisedPlaintextPort
)
{
init(
@@ -150,7 +184,8 @@ public class DruidNode
tlsPort,
enablePlaintextPort == null || enablePlaintextPort.booleanValue(),
enableTlsPort,
- labels
+ labels,
+ advertisedPlaintextPort
);
}
@@ -162,7 +197,8 @@ public class DruidNode
Integer tlsPort,
boolean enablePlaintextPort,
boolean enableTlsPort,
- Map<String, String> labels
+ Map<String, String> labels,
+ Integer advertisedPlaintextPort
)
{
Preconditions.checkNotNull(serviceName);
@@ -210,8 +246,12 @@ public class DruidNode
}
}
this.plaintextPort = plainTextPort;
+ this.advertisedPlaintextPort = advertisedPlaintextPort != null &&
advertisedPlaintextPort > 0
+ ? advertisedPlaintextPort
+ : this.plaintextPort;
} else {
this.plaintextPort = -1;
+ this.advertisedPlaintextPort = -1;
}
if (enableTlsPort) {
this.tlsPort = tlsPort;
@@ -246,6 +286,9 @@ public class DruidNode
return bindOnHost;
}
+ /**
+ * Port Jetty binds to. Peers should use {@link
#getAdvertisedPlaintextPort()} or {@link #getPortToUse()}.
+ */
public int getPlaintextPort()
{
return plaintextPort;
@@ -276,9 +319,17 @@ public class DruidNode
return buildRevision;
}
+ /**
+ * Plaintext port peers use to reach this node; falls back to {@link
#getPlaintextPort()} when unset.
+ */
+ public int getAdvertisedPlaintextPort()
+ {
+ return advertisedPlaintextPort;
+ }
+
public DruidNode withService(String service)
{
- return new DruidNode(service, host, bindOnHost, plaintextPort, tlsPort,
enablePlaintextPort, enableTlsPort);
+ return new DruidNode(service, host, bindOnHost, plaintextPort, null,
tlsPort, enablePlaintextPort, enableTlsPort, labels, advertisedPlaintextPort);
}
public String getServiceScheme()
@@ -292,10 +343,10 @@ public class DruidNode
public String getHostAndPort()
{
if (enablePlaintextPort) {
- if (plaintextPort < 0) {
+ if (advertisedPlaintextPort < 0) {
return HostAndPort.fromString(host).toString();
} else {
- return HostAndPort.fromParts(host, plaintextPort).toString();
+ return HostAndPort.fromParts(host, advertisedPlaintextPort).toString();
}
}
return null;
@@ -314,7 +365,7 @@ public class DruidNode
if (enableTlsPort) {
return getTlsPort();
} else {
- return getPlaintextPort();
+ return getAdvertisedPlaintextPort();
}
}
@@ -359,6 +410,7 @@ public class DruidNode
enablePlaintextPort == druidNode.enablePlaintextPort &&
tlsPort == druidNode.tlsPort &&
enableTlsPort == druidNode.enableTlsPort &&
+ advertisedPlaintextPort == druidNode.advertisedPlaintextPort &&
Objects.equals(serviceName, druidNode.serviceName) &&
Objects.equals(host, druidNode.host) &&
Objects.equals(labels, druidNode.labels);
@@ -367,7 +419,7 @@ public class DruidNode
@Override
public int hashCode()
{
- return Objects.hash(serviceName, host, port, plaintextPort,
enablePlaintextPort, tlsPort, enableTlsPort, labels);
+ return Objects.hash(serviceName, host, port, plaintextPort,
enablePlaintextPort, tlsPort, enableTlsPort, labels, advertisedPlaintextPort);
}
@Override
@@ -379,6 +431,7 @@ public class DruidNode
", bindOnHost=" + bindOnHost +
", port=" + port +
", plaintextPort=" + plaintextPort +
+ ", advertisedPlaintextPort=" + advertisedPlaintextPort +
", enablePlaintextPort=" + enablePlaintextPort +
", tlsPort=" + tlsPort +
", enableTlsPort=" + enableTlsPort +
diff --git
a/server/src/main/java/org/apache/druid/server/http/BaseDynamicConfigSyncer.java
b/server/src/main/java/org/apache/druid/server/http/BaseDynamicConfigSyncer.java
index 9c74298e34e..e863db2d37d 100644
---
a/server/src/main/java/org/apache/druid/server/http/BaseDynamicConfigSyncer.java
+++
b/server/src/main/java/org/apache/druid/server/http/BaseDynamicConfigSyncer.java
@@ -262,12 +262,7 @@ public abstract class
BaseDynamicConfigSyncer<DynamicConfig>
return null;
}
- return new ServiceLocation(
- druidNode.getHost(),
- druidNode.getPlaintextPort(),
- druidNode.getTlsPort(),
- ""
- );
+ return ServiceLocation.fromDruidNode(druidNode);
}
private void emitStat(CoordinatorStat stat, RowKey rowKey, long value)
diff --git a/server/src/test/java/org/apache/druid/rpc/ServiceLocationTest.java
b/server/src/test/java/org/apache/druid/rpc/ServiceLocationTest.java
index b5b8af06830..242a1ed363d 100644
--- a/server/src/test/java/org/apache/druid/rpc/ServiceLocationTest.java
+++ b/server/src/test/java/org/apache/druid/rpc/ServiceLocationTest.java
@@ -20,6 +20,7 @@
package org.apache.druid.rpc;
import nl.jqno.equalsverifier.EqualsVerifier;
+import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.DruidServerMetadata;
import org.apache.druid.server.coordination.ServerType;
import org.junit.jupiter.api.Assertions;
@@ -107,6 +108,23 @@ public class ServiceLocationTest
);
}
+ @Test
+ public void test_fromDruidNode_usesAdvertisedPlaintextPort()
+ {
+ Assertions.assertEquals(
+ new ServiceLocation("host", 8080, -1, ""),
+ ServiceLocation.fromDruidNode(new DruidNode("svc", "host", false,
8080, null, true, false))
+ );
+ Assertions.assertEquals(
+ new ServiceLocation("host", 9080, -1, ""),
+ ServiceLocation.fromDruidNode(new DruidNode("svc", "host", false,
8080, null, null, true, false, null, 9080))
+ );
+ Assertions.assertEquals(
+ new ServiceLocation("host", 9080, 8443, ""),
+ ServiceLocation.fromDruidNode(new DruidNode("svc", "host", false,
8080, null, 8443, true, true, null, 9080))
+ );
+ }
+
@Test
public void test_equals()
{
diff --git a/server/src/test/java/org/apache/druid/server/DruidNodeTest.java
b/server/src/test/java/org/apache/druid/server/DruidNodeTest.java
index 8da3a7f077d..ef4e21ac72a 100644
--- a/server/src/test/java/org/apache/druid/server/DruidNodeTest.java
+++ b/server/src/test/java/org/apache/druid/server/DruidNodeTest.java
@@ -27,6 +27,7 @@ import org.apache.druid.jackson.DefaultObjectMapper;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.net.URI;
import java.util.Map;
public class DruidNodeTest
@@ -188,6 +189,130 @@ public class DruidNodeTest
Assertions.assertEquals(-1, node.getTlsPort());
}
+ @Test
+ public void testAdvertisedPlaintextPort()
+ {
+ // When not set, advertisedPlaintextPort defaults to plaintextPort
+ DruidNode node = new DruidNode("test", "host", false, 8082, null, true,
false);
+ Assertions.assertEquals(8082, node.getPlaintextPort());
+ Assertions.assertEquals(8082, node.getAdvertisedPlaintextPort());
+ Assertions.assertEquals("host:8082", node.getHostAndPort());
+ Assertions.assertEquals("host:8082", node.getHostAndPortToUse());
+ Assertions.assertEquals(8082, node.getPortToUse());
+
+ // When set, peer-facing accessors use advertisedPlaintextPort;
getPlaintextPort() stays the bind port
+ node = new DruidNode("test", "host", false, 8082, null, 9443, true, false,
null, 9443);
+ Assertions.assertEquals(8082, node.getPlaintextPort());
+ Assertions.assertEquals(9443, node.getAdvertisedPlaintextPort());
+ Assertions.assertEquals("host:9443", node.getHostAndPort());
+ Assertions.assertEquals("host:9443", node.getHostAndPortToUse());
+ Assertions.assertEquals(9443, node.getPortToUse());
+ Assertions.assertEquals(URI.create("http://host:9443";),
node.getUriToUse());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortNonPositiveFallsBack()
+ {
+ for (Integer advertised : new Integer[]{null, 0, -1}) {
+ DruidNode node = new DruidNode("test", "host", false, 8082, null, null,
true, false, null, advertised);
+ Assertions.assertEquals(8082, node.getAdvertisedPlaintextPort());
+ Assertions.assertEquals(8082, node.getPortToUse());
+ Assertions.assertEquals("host:8082", node.getHostAndPortToUse());
+ }
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortWithTls()
+ {
+ // TLS port still wins for the *ToUse accessors when TLS is enabled
+ DruidNode node = new DruidNode("test", "host", false, 8082, null, 8443,
true, true, null, 9443);
+ Assertions.assertEquals(8082, node.getPlaintextPort());
+ Assertions.assertEquals(9443, node.getAdvertisedPlaintextPort());
+ Assertions.assertEquals(8443, node.getTlsPort());
+ Assertions.assertEquals("host:9443", node.getHostAndPort());
+ Assertions.assertEquals("host:8443", node.getHostAndTlsPort());
+ Assertions.assertEquals("host:8443", node.getHostAndPortToUse());
+ Assertions.assertEquals(8443, node.getPortToUse());
+ Assertions.assertEquals(URI.create("https://host:8443";),
node.getUriToUse());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortDisabledPlaintext()
+ {
+ DruidNode node = new DruidNode("test", "host", false, null, null, 8443,
false, true, null, 9443);
+ Assertions.assertEquals(-1, node.getPlaintextPort());
+ Assertions.assertEquals(-1, node.getAdvertisedPlaintextPort());
+ Assertions.assertNull(node.getHostAndPort());
+ Assertions.assertEquals(8443, node.getPortToUse());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortSerde() throws Exception
+ {
+ DruidNode original = new DruidNode("service", "host", true, 8082, null,
5678, true, true, null, 9443);
+ DruidNode actual = mapper.readValue(mapper.writeValueAsString(original),
DruidNode.class);
+ Assertions.assertEquals(8082, actual.getPlaintextPort());
+ Assertions.assertEquals(9443, actual.getAdvertisedPlaintextPort());
+ Assertions.assertEquals(5678, actual.getTlsPort());
+ Assertions.assertEquals("host:9443", actual.getHostAndPort());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortBackwardCompatDeserialization()
throws Exception
+ {
+ String json = "{\n"
+ + " \"service\":\"service\",\n"
+ + " \"host\":\"host\",\n"
+ + " \"plaintextPort\":8082,\n"
+ + " \"enablePlaintextPort\":true,\n"
+ + " \"enableTlsPort\":false\n"
+ + "}\n";
+ DruidNode actual = mapper.readValue(json, DruidNode.class);
+ Assertions.assertEquals(8082, actual.getPlaintextPort());
+ Assertions.assertEquals(8082, actual.getAdvertisedPlaintextPort());
+ Assertions.assertEquals("host:8082", actual.getHostAndPort());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortDeserialization() throws Exception
+ {
+ String json = "{\n"
+ + " \"service\":\"service\",\n"
+ + " \"host\":\"host\",\n"
+ + " \"plaintextPort\":8082,\n"
+ + " \"advertisedPlaintextPort\":9443,\n"
+ + " \"enablePlaintextPort\":true,\n"
+ + " \"enableTlsPort\":false\n"
+ + "}\n";
+ DruidNode actual = mapper.readValue(json, DruidNode.class);
+ Assertions.assertEquals(8082, actual.getPlaintextPort());
+ Assertions.assertEquals(9443, actual.getAdvertisedPlaintextPort());
+ Assertions.assertEquals("host:9443", actual.getHostAndPort());
+ Assertions.assertEquals("host:9443", actual.getHostAndPortToUse());
+ Assertions.assertEquals(9443, actual.getPortToUse());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortWithService()
+ {
+ DruidNode node = new DruidNode("test", "host", false, 8082, null, 9443,
true, false, null, 9443);
+ DruidNode copy = node.withService("other");
+ Assertions.assertEquals("other", copy.getServiceName());
+ Assertions.assertEquals(8082, copy.getPlaintextPort());
+ Assertions.assertEquals(9443, copy.getAdvertisedPlaintextPort());
+ }
+
+ @Test
+ public void testAdvertisedPlaintextPortEquality()
+ {
+ DruidNode a = new DruidNode("test", "host", false, 8082, null, 9443, true,
false, null, 9443);
+ DruidNode b = new DruidNode("test", "host", false, 8082, null, 9443, true,
false, null, 9443);
+ DruidNode c = new DruidNode("test", "host", false, 8082, null, true,
false);
+ Assertions.assertEquals(a, b);
+ Assertions.assertNotEquals(a, c);
+ Assertions.assertEquals(a.hashCode(), b.hashCode());
+ }
+
@Test
public void testConflictingPorts()
{
diff --git
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigSyncerTest.java
b/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigSyncerTest.java
index 4bc33131415..db3ab68d7ac 100644
---
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigSyncerTest.java
+++
b/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigSyncerTest.java
@@ -22,6 +22,7 @@ package org.apache.druid.server.http;
import com.google.common.util.concurrent.Futures;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.discovery.DiscoveryDruidNode;
import org.apache.druid.discovery.DruidNodeDiscovery;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
@@ -31,14 +32,18 @@ import
org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
import org.apache.druid.rpc.RequestBuilder;
import org.apache.druid.rpc.ServiceClient;
+import org.apache.druid.rpc.ServiceLocation;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordinator.CoordinatorConfigManager;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentMatchers;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
@@ -51,6 +56,7 @@ public class CoordinatorDynamicConfigSyncerTest
private ServiceClient serviceClient;
private CoordinatorConfigManager coordinatorConfigManager;
+ private DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
private DruidNodeDiscovery druidNodeDiscovery;
@BeforeEach
@@ -64,15 +70,15 @@ public class CoordinatorDynamicConfigSyncerTest
.when(serviceClient).asyncRequest(ArgumentMatchers.any(),
ArgumentMatchers.any());
coordinatorConfigManager = mock(CoordinatorConfigManager.class);
- DruidNodeDiscoveryProvider provider =
mock(DruidNodeDiscoveryProvider.class);
+ druidNodeDiscoveryProvider = mock(DruidNodeDiscoveryProvider.class);
druidNodeDiscovery = mock(DruidNodeDiscovery.class);
-
doReturn(druidNodeDiscovery).when(provider).getForNodeRole(NodeRole.BROKER);
+
doReturn(druidNodeDiscovery).when(druidNodeDiscoveryProvider).getForNodeRole(NodeRole.BROKER);
target = new CoordinatorDynamicConfigSyncer(
(serviceName, serviceLocator, retryPolicy) -> serviceClient,
coordinatorConfigManager,
DefaultObjectMapper.INSTANCE,
- provider,
+ druidNodeDiscoveryProvider,
mock(ServiceEmitter.class)
);
}
@@ -104,6 +110,36 @@ public class CoordinatorDynamicConfigSyncerTest
verify(serviceClient).asyncRequest(eq(requestBuilder),
ArgumentMatchers.any());
}
+ @Test
+ public void testSync_usesAdvertisedPlaintextPort()
+ {
+ final List<Set<ServiceLocation>> dialedLocations = new ArrayList<>();
+ target = new CoordinatorDynamicConfigSyncer(
+ (serviceName, serviceLocator, retryPolicy) -> {
+
dialedLocations.add(FutureUtils.getUnchecked(serviceLocator.locate(),
true).getLocations());
+ return serviceClient;
+ },
+ coordinatorConfigManager,
+ DefaultObjectMapper.INSTANCE,
+ druidNodeDiscoveryProvider,
+ mock(ServiceEmitter.class)
+ );
+
+
doReturn(CoordinatorDynamicConfig.builder().build()).when(coordinatorConfigManager).getCurrentDynamicConfig();
+ List<DiscoveryDruidNode> nodes = List.of(
+ new DiscoveryDruidNode(
+ new DruidNode("service", "host", false, 8080, null, null, true,
false, null, 9080),
+ NodeRole.BROKER,
+ null,
+ null
+ )
+ );
+ doReturn(nodes).when(druidNodeDiscovery).getAllNodes();
+
+ target.broadcastConfigToBrokers();
+ Assertions.assertEquals(List.of(Set.of(new ServiceLocation("host", 9080,
-1, ""))), dialedLocations);
+ }
+
@Test
public void testSync_whenDruidNode_isNull()
{
diff --git
a/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
b/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
index abd903a4847..9ddcbdef2e5 100644
---
a/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
+++
b/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
@@ -75,7 +75,7 @@ public class TieredBrokerHostSelectorTest
druidNodeDiscoveryProvider =
EasyMock.createStrictMock(DruidNodeDiscoveryProvider.class);
node1 = new DiscoveryDruidNode(
- new DruidNode("hotBroker", "hotHost", false, 8080, null, true, false),
+ new DruidNode("hotBroker", "hotHost", false, 8080, null, null, true,
false, null, 9080),
NodeRole.BROKER,
ImmutableMap.of()
);
@@ -192,7 +192,8 @@ public class TieredBrokerHostSelectorTest
);
Assertions.assertEquals("hotBroker", p.lhs);
- Assertions.assertEquals("hotHost:8080", p.rhs.getHost());
+ Assertions.assertEquals("hotHost:9080", p.rhs.getHost());
+ Assertions.assertEquals(9080, p.rhs.getPort());
}
@Test
@@ -367,7 +368,7 @@ public class TieredBrokerHostSelectorTest
ImmutableMap.of(
"mediumBroker", ImmutableList.of(),
"coldBroker", ImmutableList.of("coldHost1:8080", "coldHost2:8080"),
- "hotBroker", ImmutableList.of("hotHost:8080")
+ "hotBroker", ImmutableList.of("hotHost:9080")
),
Maps.transformValues(
brokerSelector.getAllBrokers(),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
