You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/01/14 10:56:18 UTC
[pulsar] branch master updated: [fix][broker] Support zookeeper read-only config. (#19156)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new accae9f371f [fix][broker] Support zookeeper read-only config. (#19156)
accae9f371f is described below
commit accae9f371f87dcf0b2a2f8d4ed277b742bfdc5f
Author: Yan Zhao <ho...@apache.org>
AuthorDate: Sat Jan 14 18:56:05 2023 +0800
[fix][broker] Support zookeeper read-only config. (#19156)
---
.../java/org/apache/pulsar/broker/ServiceConfiguration.java | 6 ++++++
.../org/apache/pulsar/broker/resources/PulsarResources.java | 10 ++++++----
.../src/main/java/org/apache/pulsar/broker/PulsarService.java | 4 ++--
.../apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java | 4 +++-
.../apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java | 4 +++-
.../apache/pulsar/websocket/proxy/ProxyConfigurationTest.java | 4 +++-
.../websocket/proxy/ProxyEncryptionPublishConsumeTest.java | 4 +++-
.../apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java | 4 +++-
.../pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java | 4 +++-
.../websocket/proxy/ProxyPublishConsumeWithoutZKTest.java | 4 +++-
.../pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java | 4 +++-
.../java/org/apache/pulsar/functions/worker/WorkerConfig.java | 6 ++++++
.../main/java/org/apache/pulsar/functions/worker/Worker.java | 3 ++-
.../org/apache/pulsar/proxy/server/ProxyConfiguration.java | 6 ++++++
.../main/java/org/apache/pulsar/proxy/server/ProxyService.java | 6 ++++--
.../java/org/apache/pulsar/websocket/WebSocketService.java | 8 +++++---
16 files changed, 61 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index a39ebff49a0..1f27641b1fa 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -461,6 +461,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private int zooKeeperCacheExpirySeconds = -1;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Is zookeeper allow read-only operations."
+ )
+ private boolean zooKeeperAllowReadOnlyOperations;
+
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
index 161a44703a7..dfcd0a4194f 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
@@ -87,17 +87,19 @@ public class PulsarResources {
this.configurationMetadataStore = Optional.ofNullable(configurationMetadataStore);
}
- public static MetadataStoreExtended createLocalMetadataStore(String serverUrls, int sessionTimeoutMs)
+ public static MetadataStoreExtended createLocalMetadataStore(String serverUrls, int sessionTimeoutMs,
+ boolean allowReadOnlyOperations)
throws MetadataStoreException {
return MetadataStoreExtended.create(serverUrls, MetadataStoreConfig.builder()
- .sessionTimeoutMillis(sessionTimeoutMs).allowReadOnlyOperations(false)
+ .sessionTimeoutMillis(sessionTimeoutMs).allowReadOnlyOperations(allowReadOnlyOperations)
.metadataStoreName(MetadataStoreConfig.METADATA_STORE).build());
}
- public static MetadataStoreExtended createConfigMetadataStore(String serverUrls, int sessionTimeoutMs)
+ public static MetadataStoreExtended createConfigMetadataStore(String serverUrls, int sessionTimeoutMs,
+ boolean allowReadOnlyOperations)
throws MetadataStoreException {
return MetadataStoreExtended.create(serverUrls, MetadataStoreConfig.builder()
- .sessionTimeoutMillis(sessionTimeoutMs).allowReadOnlyOperations(false)
+ .sessionTimeoutMillis(sessionTimeoutMs).allowReadOnlyOperations(allowReadOnlyOperations)
.metadataStoreName(MetadataStoreConfig.CONFIGURATION_METADATA_STORE).build());
}
}
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 3a76842289d..742f34c8d0e 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
@@ -356,7 +356,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return MetadataStoreFactory.create(config.getConfigurationMetadataStoreUrl(),
MetadataStoreConfig.builder()
.sessionTimeoutMillis((int) config.getMetadataStoreSessionTimeoutMillis())
- .allowReadOnlyOperations(false)
+ .allowReadOnlyOperations(config.isZooKeeperAllowReadOnlyOperations())
.configFilePath(config.getMetadataStoreConfigPath())
.batchingEnabled(config.isMetadataStoreBatchingEnabled())
.batchingMaxDelayMillis(config.getMetadataStoreBatchingMaxDelayMillis())
@@ -1063,7 +1063,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return MetadataStoreExtended.create(config.getMetadataStoreUrl(),
MetadataStoreConfig.builder()
.sessionTimeoutMillis((int) config.getMetadataStoreSessionTimeoutMillis())
- .allowReadOnlyOperations(false)
+ .allowReadOnlyOperations(config.isZooKeeperAllowReadOnlyOperations())
.configFilePath(config.getMetadataStoreConfigPath())
.batchingEnabled(config.isMetadataStoreBatchingEnabled())
.batchingMaxDelayMillis(config.getMetadataStoreBatchingMaxDelayMillis())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
index be8076dcd50..e4d7b8349ec 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -82,7 +83,8 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
}
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
index 5f8d4e01c81..a3b26a4a9d1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -64,7 +65,8 @@ public class ProxyAuthorizationTest extends MockedPulsarServiceBaseTest {
config.setWebServicePort(Optional.of(0));
config.setConfigurationMetadataStoreUrl(GLOBAL_DUMMY_VALUE);
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
service.start();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
index 0b85cbce516..173948ab1be 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -67,7 +68,8 @@ public class ProxyConfigurationTest extends ProducerConsumerBase {
config.setServiceUrl("http://localhost:8080");
config.getProperties().setProperty("brokerClient_lookupTimeoutMs", "100");
WebSocketService service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
service.start();
PulsarClientImpl client = (PulsarClientImpl) service.getPulsarClient();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyEncryptionPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyEncryptionPublishConsumeTest.java
index e9d18f4d401..041107e5e93 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyEncryptionPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyEncryptionPublishConsumeTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.websocket.proxy;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -73,7 +74,8 @@ public class ProxyEncryptionPublishConsumeTest extends ProducerConsumerBase {
config.setConfigurationMetadataStoreUrl(GLOBAL_DUMMY_VALUE);
config.setCryptoKeyReaderFactoryClassName(CryptoKeyReaderFactoryImpl.class.getName());
WebSocketService service = spy(new WebSocketService(config));
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 4fe6b7c0ab8..8c64e40f927 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -99,7 +100,8 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
config.setClusterName("test");
config.setConfigurationMetadataStoreUrl(GLOBAL_DUMMY_VALUE);
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
index 468e97fcc0f..3ee9b6127de 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -72,7 +73,8 @@ public class ProxyPublishConsumeTlsTest extends TlsProducerConsumerBase {
config.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
config.setConfigurationMetadataStoreUrl(GLOBAL_DUMMY_VALUE);
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
index 5d4668cf7cd..0a432406001 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -58,7 +59,8 @@ public class ProxyPublishConsumeWithoutZKTest extends ProducerConsumerBase {
config.setServiceUrl(pulsar.getSafeWebServiceAddress());
config.setServiceUrlTls(pulsar.getWebServiceAddressTls());
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeper)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeper)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
index d0c0fcbd6da..01c851290b6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.proxy.v1;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -83,7 +84,8 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
}
service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
- doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createConfigMetadataStore(anyString(), anyInt());
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 876f62bba2c..c7f2e116f5b 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -224,6 +224,12 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
)
private int zooKeeperCacheExpirySeconds = -1;
+ @FieldContext(
+ category = CATEGORY_WORKER,
+ doc = "Is zooKeeper allow read-only operations."
+ )
+ private boolean zooKeeperAllowReadOnlyOperations;
+
@FieldContext(
category = CATEGORY_CONNECTORS,
doc = "The path to the location to locate builtin connectors"
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
index 03a04036999..cfb71b9d429 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Worker.java
@@ -76,7 +76,8 @@ public class Worker {
try {
configMetadataStore = PulsarResources.createConfigMetadataStore(
workerConfig.getConfigurationMetadataStoreUrl(),
- (int) workerConfig.getMetadataStoreSessionTimeoutMillis());
+ (int) workerConfig.getMetadataStoreSessionTimeoutMillis(),
+ workerConfig.isZooKeeperAllowReadOnlyOperations());
} catch (IOException e) {
throw new PulsarServerException(e);
}
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
index 6162398b75a..f2d9d9451dc 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
@@ -157,6 +157,12 @@ public class ProxyConfiguration implements PulsarConfiguration {
)
private int zooKeeperCacheExpirySeconds = -1;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Is zooKeeper allow read-only operations."
+ )
+ private boolean zooKeeperAllowReadOnlyOperations;
+
@FieldContext(
category = CATEGORY_BROKER_DISCOVERY,
doc = "The service url points to the broker cluster. URL must have the pulsar:// prefix."
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
index db061fe9c59..01f02efb96c 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
@@ -448,12 +448,14 @@ public class ProxyService implements Closeable {
public MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
return PulsarResources.createLocalMetadataStore(proxyConfig.getMetadataStoreUrl(),
- proxyConfig.getMetadataStoreSessionTimeoutMillis());
+ proxyConfig.getMetadataStoreSessionTimeoutMillis(),
+ proxyConfig.isZooKeeperAllowReadOnlyOperations());
}
public MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
return PulsarResources.createConfigMetadataStore(proxyConfig.getConfigurationMetadataStoreUrl(),
- proxyConfig.getMetadataStoreSessionTimeoutMillis());
+ proxyConfig.getMetadataStoreSessionTimeoutMillis(),
+ proxyConfig.isZooKeeperAllowReadOnlyOperations());
}
public Authentication getProxyClientAuthenticationPlugin() {
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 f43c36c051f..04a9354b7a5 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
@@ -107,7 +107,8 @@ public class WebSocketService implements Closeable {
if (isNotBlank(config.getConfigurationMetadataStoreUrl())) {
try {
configMetadataStore = createConfigMetadataStore(config.getConfigurationMetadataStoreUrl(),
- (int) config.getMetadataStoreSessionTimeoutMillis());
+ (int) config.getMetadataStoreSessionTimeoutMillis(),
+ config.isZooKeeperAllowReadOnlyOperations());
} catch (MetadataStoreException e) {
throw new PulsarServerException(e);
}
@@ -140,9 +141,10 @@ public class WebSocketService implements Closeable {
log.info("Pulsar WebSocket Service started");
}
- public MetadataStoreExtended createConfigMetadataStore(String serverUrls, int sessionTimeoutMs)
+ public MetadataStoreExtended createConfigMetadataStore(String serverUrls, int sessionTimeoutMs, boolean
+ isAllowReadOnlyOperations)
throws MetadataStoreException {
- return PulsarResources.createConfigMetadataStore(serverUrls, sessionTimeoutMs);
+ return PulsarResources.createConfigMetadataStore(serverUrls, sessionTimeoutMs, isAllowReadOnlyOperations);
}
@Override