[improve] [broker] Introduce metadataStoreAllowReadOnlyOperations to deprecate zookeeperAllowReadOnlyOperations (#19246)

horizonzy · web-flow · commit 9bb3a213e8d7 · 2023-01-18T17:18:20.000+08:00
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
@@ -434,6 +434,12 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private int metadataStoreCacheExpirySeconds = 300;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Is metadata store read-only operations."
+    )
+    private boolean metadataStoreAllowReadOnlyOperations;
+
     @Deprecated
     @FieldContext(
         category = CATEGORY_SERVER,
@@ -461,8 +467,10 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
         )
     private int zooKeeperCacheExpirySeconds = -1;
 
+    @Deprecated
     @FieldContext(
             category = CATEGORY_SERVER,
+            deprecated = true,
             doc = "Is zookeeper allow read-only operations."
     )
     private boolean zooKeeperAllowReadOnlyOperations;
@@ -3249,6 +3257,10 @@ public int getMetadataStoreCacheExpirySeconds() {
         return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds;
     }
 
+    public boolean isMetadataStoreAllowReadOnlyOperations() {
+        return zooKeeperAllowReadOnlyOperations || metadataStoreAllowReadOnlyOperations;
+    }
+
     public long getManagedLedgerCacheEvictionIntervalMs() {
         return managedLedgerCacheEvictionFrequency > 0
                 ? (long) (1000 / Math.max(
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
@@ -356,7 +356,7 @@ public MetadataStore createConfigurationMetadataStore(PulsarMetadataEventSynchro
         return MetadataStoreFactory.create(config.getConfigurationMetadataStoreUrl(),
                 MetadataStoreConfig.builder()
                         .sessionTimeoutMillis((int) config.getMetadataStoreSessionTimeoutMillis())
-                        .allowReadOnlyOperations(config.isZooKeeperAllowReadOnlyOperations())
+                        .allowReadOnlyOperations(config.isMetadataStoreAllowReadOnlyOperations())
                         .configFilePath(config.getMetadataStoreConfigPath())
                         .batchingEnabled(config.isMetadataStoreBatchingEnabled())
                         .batchingMaxDelayMillis(config.getMetadataStoreBatchingMaxDelayMillis())
@@ -1063,7 +1063,7 @@ public MetadataStoreExtended createLocalMetadataStore(PulsarMetadataEventSynchro
         return MetadataStoreExtended.create(config.getMetadataStoreUrl(),
                 MetadataStoreConfig.builder()
                         .sessionTimeoutMillis((int) config.getMetadataStoreSessionTimeoutMillis())
-                        .allowReadOnlyOperations(config.isZooKeeperAllowReadOnlyOperations())
+                        .allowReadOnlyOperations(config.isMetadataStoreAllowReadOnlyOperations())
                         .configFilePath(config.getMetadataStoreConfigPath())
                         .batchingEnabled(config.isMetadataStoreBatchingEnabled())
                         .batchingMaxDelayMillis(config.getMetadataStoreBatchingMaxDelayMillis())
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
@@ -197,6 +197,12 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
     )
     private int metadataStoreCacheExpirySeconds = 300;
 
+    @FieldContext(
+            category = CATEGORY_WORKER,
+            doc = "Is metadata store read-only operations."
+    )
+    private boolean metadataStoreAllowReadOnlyOperations;
+
     @Deprecated
     @FieldContext(
             category = CATEGORY_WORKER,
@@ -224,8 +230,10 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
     )
     private int zooKeeperCacheExpirySeconds = -1;
 
+    @Deprecated
     @FieldContext(
             category = CATEGORY_WORKER,
+            deprecated = true,
             doc = "Is zooKeeper allow read-only operations."
     )
     private boolean zooKeeperAllowReadOnlyOperations;
@@ -911,4 +919,8 @@ public int getMetadataStoreOperationTimeoutSeconds() {
     public int getMetadataStoreCacheExpirySeconds() {
         return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds;
     }
+
+    public boolean isMetadataStoreAllowReadOnlyOperations() {
+        return zooKeeperAllowReadOnlyOperations || metadataStoreAllowReadOnlyOperations;
+    }
 }
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
@@ -77,7 +77,7 @@ private AuthorizationService getAuthorizationService() throws PulsarServerExcept
                 configMetadataStore = PulsarResources.createConfigMetadataStore(
                         workerConfig.getConfigurationMetadataStoreUrl(),
                         (int) workerConfig.getMetadataStoreSessionTimeoutMillis(),
-                        workerConfig.isZooKeeperAllowReadOnlyOperations());
+                        workerConfig.isMetadataStoreAllowReadOnlyOperations());
             } 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
@@ -133,6 +133,12 @@ public class ProxyConfiguration implements PulsarConfiguration {
     )
     private int metadataStoreCacheExpirySeconds = 300;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Is metadata store read-only operations."
+    )
+    private boolean metadataStoreAllowReadOnlyOperations;
+
     @FieldContext(
             category = CATEGORY_SERVER,
             doc = "Max size of messages.",
@@ -157,8 +163,10 @@ public class ProxyConfiguration implements PulsarConfiguration {
     )
     private int zooKeeperCacheExpirySeconds = -1;
 
+    @Deprecated
     @FieldContext(
             category = CATEGORY_SERVER,
+            deprecated = true,
             doc = "Is zooKeeper allow read-only operations."
     )
     private boolean zooKeeperAllowReadOnlyOperations;
@@ -916,4 +924,8 @@ public int getMetadataStoreSessionTimeoutMillis() {
     public int getMetadataStoreCacheExpirySeconds() {
         return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds;
     }
+
+    public boolean isMetadataStoreAllowReadOnlyOperations() {
+        return zooKeeperAllowReadOnlyOperations || metadataStoreAllowReadOnlyOperations;
+    }
 }
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
@@ -449,13 +449,13 @@ public Optional<Integer> getListenPortTls() {
     public MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
         return PulsarResources.createLocalMetadataStore(proxyConfig.getMetadataStoreUrl(),
                 proxyConfig.getMetadataStoreSessionTimeoutMillis(),
-                proxyConfig.isZooKeeperAllowReadOnlyOperations());
+                proxyConfig.isMetadataStoreAllowReadOnlyOperations());
     }
 
     public MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
         return PulsarResources.createConfigMetadataStore(proxyConfig.getConfigurationMetadataStoreUrl(),
                 proxyConfig.getMetadataStoreSessionTimeoutMillis(),
-                proxyConfig.isZooKeeperAllowReadOnlyOperations());
+                proxyConfig.isMetadataStoreAllowReadOnlyOperations());
     }
 
     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
@@ -113,7 +113,7 @@ public void start() throws PulsarServerException, PulsarClientException, Malform
             try {
                 configMetadataStore = createConfigMetadataStore(config.getConfigurationMetadataStoreUrl(),
                         (int) config.getMetadataStoreSessionTimeoutMillis(),
-                        config.isZooKeeperAllowReadOnlyOperations());
+                        config.isMetadataStoreAllowReadOnlyOperations());
             } catch (MetadataStoreException e) {
                 throw new PulsarServerException(e);
             }

Original file line number	Diff line number	Diff line change
`@@ -197,6 +197,12 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {`
`197`	`197`	`)`
`198`	`198`	`private int metadataStoreCacheExpirySeconds = 300;`
`199`	`199`
	`200`	`+ @FieldContext(`
	`201`	`+ category = CATEGORY_WORKER,`
	`202`	`+ doc = "Is metadata store read-only operations."`
	`203`	`+ )`
	`204`	`+ private boolean metadataStoreAllowReadOnlyOperations;`
	`205`	`+`
`200`	`206`	`@Deprecated`
`201`	`207`	`@FieldContext(`
`202`	`208`	`category = CATEGORY_WORKER,`
`@@ -224,8 +230,10 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {`
`224`	`230`	`)`
`225`	`231`	`private int zooKeeperCacheExpirySeconds = -1;`
`226`	`232`
	`233`	`+ @Deprecated`
`227`	`234`	`@FieldContext(`
`228`	`235`	`category = CATEGORY_WORKER,`
	`236`	`+ deprecated = true,`
`229`	`237`	`doc = "Is zooKeeper allow read-only operations."`
`230`	`238`	`)`
`231`	`239`	`private boolean zooKeeperAllowReadOnlyOperations;`
`@@ -911,4 +919,8 @@ public int getMetadataStoreOperationTimeoutSeconds() {`
`911`	`919`	`public int getMetadataStoreCacheExpirySeconds() {`
`912`	`920`	`return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds;`
`913`	`921`	`}`
	`922`	`+`
	`923`	`+ public boolean isMetadataStoreAllowReadOnlyOperations() {`
	`924`	`+ return zooKeeperAllowReadOnlyOperations \|\| metadataStoreAllowReadOnlyOperations;`
	`925`	`+ }`
`914`	`926`	`}`
Original file line number	Diff line number	Diff line change
`@@ -77,7 +77,7 @@ private AuthorizationService getAuthorizationService() throws PulsarServerExcept`
`77`	`77`	`configMetadataStore = PulsarResources.createConfigMetadataStore(`
`78`	`78`	`workerConfig.getConfigurationMetadataStoreUrl(),`
`79`	`79`	`(int) workerConfig.getMetadataStoreSessionTimeoutMillis(),`
`80`		`- workerConfig.isZooKeeperAllowReadOnlyOperations());`
	`80`	`+ workerConfig.isMetadataStoreAllowReadOnlyOperations());`
`81`	`81`	`} catch (IOException e) {`
`82`	`82`	`throw new PulsarServerException(e);`
`83`	`83`	`}`
Original file line number	Diff line number	Diff line change
`@@ -133,6 +133,12 @@ public class ProxyConfiguration implements PulsarConfiguration {`
`133`	`133`	`)`
`134`	`134`	`private int metadataStoreCacheExpirySeconds = 300;`
`135`	`135`
	`136`	`+ @FieldContext(`
	`137`	`+ category = CATEGORY_SERVER,`
	`138`	`+ doc = "Is metadata store read-only operations."`
	`139`	`+ )`
	`140`	`+ private boolean metadataStoreAllowReadOnlyOperations;`
	`141`	`+`
`136`	`142`	`@FieldContext(`
`137`	`143`	`category = CATEGORY_SERVER,`
`138`	`144`	`doc = "Max size of messages.",`
`@@ -157,8 +163,10 @@ public class ProxyConfiguration implements PulsarConfiguration {`
`157`	`163`	`)`
`158`	`164`	`private int zooKeeperCacheExpirySeconds = -1;`
`159`	`165`
	`166`	`+ @Deprecated`
`160`	`167`	`@FieldContext(`
`161`	`168`	`category = CATEGORY_SERVER,`
	`169`	`+ deprecated = true,`
`162`	`170`	`doc = "Is zooKeeper allow read-only operations."`
`163`	`171`	`)`
`164`	`172`	`private boolean zooKeeperAllowReadOnlyOperations;`
`@@ -916,4 +924,8 @@ public int getMetadataStoreSessionTimeoutMillis() {`
`916`	`924`	`public int getMetadataStoreCacheExpirySeconds() {`
`917`	`925`	`return zooKeeperCacheExpirySeconds > 0 ? zooKeeperCacheExpirySeconds : metadataStoreCacheExpirySeconds;`
`918`	`926`	`}`
	`927`	`+`
	`928`	`+ public boolean isMetadataStoreAllowReadOnlyOperations() {`
	`929`	`+ return zooKeeperAllowReadOnlyOperations \|\| metadataStoreAllowReadOnlyOperations;`
	`930`	`+ }`
`919`	`931`	`}`
Original file line number	Diff line number	Diff line change
`@@ -449,13 +449,13 @@ public Optional<Integer> getListenPortTls() {`
`449`	`449`	`public MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {`
`450`	`450`	`return PulsarResources.createLocalMetadataStore(proxyConfig.getMetadataStoreUrl(),`
`451`	`451`	`proxyConfig.getMetadataStoreSessionTimeoutMillis(),`
`452`		`- proxyConfig.isZooKeeperAllowReadOnlyOperations());`
	`452`	`+ proxyConfig.isMetadataStoreAllowReadOnlyOperations());`
`453`	`453`	`}`
`454`	`454`
`455`	`455`	`public MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {`
`456`	`456`	`return PulsarResources.createConfigMetadataStore(proxyConfig.getConfigurationMetadataStoreUrl(),`
`457`	`457`	`proxyConfig.getMetadataStoreSessionTimeoutMillis(),`
`458`		`- proxyConfig.isZooKeeperAllowReadOnlyOperations());`
	`458`	`+ proxyConfig.isMetadataStoreAllowReadOnlyOperations());`
`459`	`459`	`}`
`460`	`460`
`461`	`461`	`public Authentication getProxyClientAuthenticationPlugin() {`
Original file line number	Diff line number	Diff line change
`@@ -113,7 +113,7 @@ public void start() throws PulsarServerException, PulsarClientException, Malform`
`113`	`113`	`try {`
`114`	`114`	`configMetadataStore = createConfigMetadataStore(config.getConfigurationMetadataStoreUrl(),`
`115`	`115`	`(int) config.getMetadataStoreSessionTimeoutMillis(),`
`116`		`- config.isZooKeeperAllowReadOnlyOperations());`
	`116`	`+ config.isMetadataStoreAllowReadOnlyOperations());`
`117`	`117`	`} catch (MetadataStoreException e) {`
`118`	`118`	`throw new PulsarServerException(e);`
`119`	`119`	`}`