You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/20 16:08:38 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 5dac7d348bbc [fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
5dac7d348bbc is described below
commit 5dac7d348bbc6678669cbe94967d9d647a6861d2
Author: Penghui Li <pe...@apache.org>
AuthorDate: Tue Oct 20 09:04:49 2026 -0700
[fix][broker] Fix the broker's WebSocket service ignoring configurationStoreConfigPath (#27015)
(cherry picked from commit de6ca7f7387d339727c4d47b8bcd5988de5b314f)
---
.../org/apache/pulsar/broker/PulsarService.java | 5 +-
.../apache/pulsar/broker/PulsarServiceTest.java | 9 +++
.../apache/pulsar/websocket/WebSocketService.java | 11 ++-
.../WebSocketTlsConfigurationStoreTest.java | 90 ++++++++++++++++++++++
.../src/test/resources/pulsar-websocket.xml | 1 +
5 files changed, 113 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 5333d48a2825..0d9e01a9b8d7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1277,8 +1277,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
ServiceConfiguration config)
throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException {
if (config.isWebSocketServiceEnabled()) {
- // Use local broker address to avoid different IP address when using a VIP for service discovery
- this.webSocketService = new WebSocketService(null, config);
+ // Use local broker address to avoid different IP address when using a VIP for service discovery.
+ // Share the broker's configuration store client, which applies settings like configurationStoreConfigPath.
+ this.webSocketService = new WebSocketService(null, config, pulsarResources);
this.webSocketService.start();
addWebSocketServlet(new WebSocketProducerServlet(webSocketService), attributeMap,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
index bf04445fbed9..8f84a0d69c92 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceTest.java
@@ -436,6 +436,15 @@ public class PulsarServiceTest extends MockedPulsarServiceBaseTest {
assertThat(expiryLogged).as("INFO expiry while shuttingDown=%s", shuttingDown).isEqualTo(!shuttingDown);
}
+ @Test
+ public void testWebSocketServiceSharesBrokerConfigurationStore() throws Exception {
+ // Only the broker's own (here mocked) client reaches the configuration store, as with a TLS store set up
+ // through configurationStoreConfigPath; a WebSocket service opening its own client fails the startup.
+ conf.setWebSocketServiceEnabled(true);
+ setup();
+ assertThat(pulsar.getWebSocketService()).as("embedded WebSocket service").isNotNull();
+ }
+
@Test
public void testMetadataSerDesThreads() throws Exception {
final var numSerDesThreads = 5;
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
index 3dff261fbca7..4d346c7dae0a 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -85,17 +85,26 @@ public class WebSocketService implements Closeable {
}
public WebSocketService(ClusterData localCluster, ServiceConfiguration config) {
+ this(localCluster, config, null);
+ }
+
+ /**
+ * @param pulsarResources resources to share, such as the broker's, so that no configuration metadata store is
+ * opened by this service; {@code null} to open one in {@link #start()}
+ */
+ public WebSocketService(ClusterData localCluster, ServiceConfiguration config, PulsarResources pulsarResources) {
this.config = config;
this.executor = Executors
.newScheduledThreadPool(config.getWebSocketNumServiceThreads(),
new DefaultThreadFactory("pulsar-websocket"));
this.localCluster = localCluster;
+ this.pulsarResources = pulsarResources;
this.proxyStats = new ProxyStats(this);
}
public void start() throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException {
- if (isNotBlank(config.getConfigurationMetadataStoreUrl())) {
+ if (pulsarResources == null && isNotBlank(config.getConfigurationMetadataStoreUrl())) {
try {
configMetadataStore = createConfigMetadataStore(config.getConfigurationMetadataStoreUrl(),
(int) config.getMetadataStoreSessionTimeoutMillis(),
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java
new file mode 100644
index 000000000000..4d7df02e99c2
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/websocket/WebSocketTlsConfigurationStoreTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.websocket;
+
+import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_HTTP_PORT;
+import static org.apache.pulsar.tests.integration.containers.PulsarContainer.CS_PORT;
+import java.util.Map;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.containers.CSContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+import org.testcontainers.images.builder.Transferable;
+import org.testng.annotations.Test;
+
+/**
+ * The broker's WebSocket service when the broker reaches the configuration store over TLS, which only the client
+ * configuration file in {@code configurationStoreConfigPath} sets up. The WebSocket service has to share the
+ * broker's configuration store client: a client of its own would connect in plaintext and fail the broker startup.
+ */
+public class WebSocketTlsConfigurationStoreTest extends WebSocketTestSuite {
+
+ private static final int CS_TLS_PORT = 2281;
+ private static final String CS_CLIENT_CONFIG_PATH = "/pulsar/conf/configuration-store-client.conf";
+
+ @Override
+ protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster(
+ String clusterName,
+ PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) {
+ specBuilder.numBrokers(1);
+ specBuilder.numProxies(0);
+ specBuilder.brokerEnvs(Map.of(
+ "webSocketServiceEnabled", "true",
+ "configurationMetadataStoreUrl", CSContainer.NAME + ":" + CS_TLS_PORT,
+ // Not in broker.conf, so it is added with the prefix
+ "PULSAR_PREFIX_configurationStoreConfigPath", CS_CLIENT_CONFIG_PATH,
+ // The test certificate is for localhost. Netty 4.2 clients check the hostname by default, which
+ // zookeeper.ssl.hostnameVerification=false doesn't turn off
+ "PULSAR_EXTRA_OPTS", "-Dio.netty.handler.ssl.defaultEndpointVerificationAlgorithm=NONE"));
+ return super.beforeSetupCluster(clusterName, specBuilder);
+ }
+
+ @Override
+ protected void beforeStartCluster() {
+ pulsarCluster.forEachContainer(container -> {
+ if (container instanceof CSContainer) {
+ // A TLS-only port for the broker, and the plaintext one for the cluster metadata initialization
+ container.withCopyToContainer(Transferable.of("""
+ tickTime=2000
+ dataDir=data/global-zookeeper
+ clientPort=%d
+ secureClientPort=%d
+ serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
+ ssl.keyStore.location=/pulsar/certificate-authority/jks/broker.keystore.jks
+ ssl.keyStore.password=111111
+ ssl.clientAuth=none
+ """.formatted(CS_PORT, CS_TLS_PORT)), "/pulsar/conf/global_zookeeper.conf");
+ } else if (container instanceof BrokerContainer) {
+ container.withCopyToContainer(Transferable.of("""
+ zookeeper.client.secure=true
+ zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
+ zookeeper.ssl.trustStore.location=/pulsar/certificate-authority/jks/broker.truststore.jks
+ zookeeper.ssl.trustStore.password=111111
+ # The certificate is for localhost, not for the configuration store's hostname
+ zookeeper.ssl.hostnameVerification=false
+ """), CS_CLIENT_CONFIG_PATH);
+ }
+ });
+ }
+
+ @Test
+ public void testBroker() throws Exception {
+ BrokerContainer broker = pulsarCluster.getAnyBroker();
+ testWebSocket("ws://" + broker.getHost() + ":" + broker.getMappedPort(BROKER_HTTP_PORT));
+ }
+}
diff --git a/tests/integration/src/test/resources/pulsar-websocket.xml b/tests/integration/src/test/resources/pulsar-websocket.xml
index 87bf832d4e40..36f0b8f19744 100644
--- a/tests/integration/src/test/resources/pulsar-websocket.xml
+++ b/tests/integration/src/test/resources/pulsar-websocket.xml
@@ -23,6 +23,7 @@
<test name="messaging-test-suite" preserve-order="true">
<classes>
<class name="org.apache.pulsar.tests.integration.websocket.TestWebSocket" />
+ <class name="org.apache.pulsar.tests.integration.websocket.WebSocketTlsConfigurationStoreTest" />
</classes>
</test>
</suite>
\ No newline at end of file