From 17cc8ffcf2ee58637d57411422db07823b08986f Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Sat, 13 Nov 2021 21:01:59 +0800
Subject: [PATCH 1/6] add isAllowAutoUpdateSchema

---
 .../pulsar/broker/ServiceConfiguration.java   |  8 ++
 .../pulsar/broker/service/AbstractTopic.java  | 29 +++----
 .../SchemaCompatibilityCheckTest.java         | 81 +++++++++++++++++++
 3 files changed, 104 insertions(+), 14 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 b832c9568d718..1efca41d9c6c6 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
@@ -573,6 +573,14 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int brokerMaxConnectionsPerIp = 0;
 
+    @FieldContext(
+        category = CATEGORY_POLICIES,
+        dynamic = true,
+        doc = "Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace "
+            + "policy will be ignored. This is enabled by default."
+    )
+    private boolean isAllowAutoUpdateSchema = true;
+
     @FieldContext(
         category = CATEGORY_SERVER,
         dynamic = true,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index f4f7615688fc2..0511c05bf255d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -333,20 +333,21 @@ public CompletableFuture<SchemaVersion> addSchema(SchemaData schema) {
         String base = TopicName.get(getName()).getPartitionedTopicName();
         String id = TopicName.get(base).getSchemaName();
         SchemaRegistryService schemaRegistryService = brokerService.pulsar().getSchemaRegistryService();
-        return isAllowAutoUpdateSchema ? schemaRegistryService
-                .putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy)
-                : schemaRegistryService.trimDeletedSchemaAndGetList(id).thenCompose(schemaAndMetadataList ->
-                schemaRegistryService.getSchemaVersionBySchemaData(schemaAndMetadataList, schema)
-                        .thenCompose(schemaVersion -> {
-                    if (schemaVersion == null) {
-                        return FutureUtil
-                                .failedFuture(
-                                        new IncompatibleSchemaException(
-                                                "Schema not found and schema auto updating is disabled."));
-                    } else {
-                        return CompletableFuture.completedFuture(schemaVersion);
-                    }
-                }));
+
+        if (brokerService.pulsar().getConfig().isAllowAutoUpdateSchema() && isAllowAutoUpdateSchema) {
+            return schemaRegistryService.putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy);
+        } else {
+            return schemaRegistryService.trimDeletedSchemaAndGetList(id).thenCompose(schemaAndMetadataList ->
+                    schemaRegistryService.getSchemaVersionBySchemaData(schemaAndMetadataList, schema)
+                            .thenCompose(schemaVersion -> {
+                                if (schemaVersion == null) {
+                                    return FutureUtil.failedFuture(new IncompatibleSchemaException(
+                                            "Schema not found and schema auto updating is disabled."));
+                                } else {
+                                    return CompletableFuture.completedFuture(schemaVersion);
+                                }
+                            }));
+        }
     }
 
     @Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
index 293f71d5f878f..b169723dc4463 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
@@ -218,6 +218,87 @@ public void testConsumerCompatibilityReadAllCheckTest(SchemaCompatibilityStrateg
         }
     }
 
+    @Test(dataProvider = "AllCheckSchemaCompatibilityStrategy")
+    public void testBrokerAllowAutoUpdateSchemaDisabled(SchemaCompatibilityStrategy schemaCompatibilityStrategy)
+            throws Exception {
+
+        final String tenant = PUBLIC_TENANT;
+        final String topic = "test-consumer-compatibility";
+        String namespace = "test-namespace-" + randomName(16);
+        String fqtn = TopicName.get(
+                TopicDomain.persistent.value(),
+                tenant,
+                namespace,
+                topic
+        ).toString();
+
+        NamespaceName namespaceName = NamespaceName.get(tenant, namespace);
+
+        admin.namespaces().createNamespace(
+                tenant + "/" + namespace,
+                Sets.newHashSet(CLUSTER_NAME)
+        );
+
+        assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
+                SchemaCompatibilityStrategy.FULL);
+
+        admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
+        admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
+
+
+        pulsar.getConfig().setAllowAutoUpdateSchema(false);
+        ProducerBuilder<Schemas.PersonTwo> producerThreeBuilder = pulsarClient
+                .newProducer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
+                                (false).withSupportSchemaVersioning(true).
+                        withPojo(Schemas.PersonTwo.class).build()))
+                .topic(fqtn);
+        try {
+            producerThreeBuilder.create();
+        } catch (Exception e) {
+            Assert.assertTrue(e.getMessage().contains("Schema not found and schema auto updating is disabled."));
+        }
+
+        pulsar.getConfig().setAllowAutoUpdateSchema(true);
+        ConsumerBuilder<Schemas.PersonTwo> comsumerBuilder = pulsarClient.newConsumer(Schema.AVRO(
+                        SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
+                                        (false).withSupportSchemaVersioning(true).
+                                withPojo(Schemas.PersonTwo.class).build()))
+                .subscriptionName("test")
+                .topic(fqtn);
+
+        Producer<Schemas.PersonTwo> producer = producerThreeBuilder.create();
+        Consumer<Schemas.PersonTwo> consumerTwo = comsumerBuilder.subscribe();
+
+        producer.send(new Schemas.PersonTwo(2, "Lucy"));
+        Message<Schemas.PersonTwo> message = consumerTwo.receive();
+
+        Schemas.PersonTwo personTwo = message.getValue();
+        consumerTwo.acknowledge(message);
+
+        assertEquals(personTwo.getId(), 2);
+        assertEquals(personTwo.getName(), "Lucy");
+
+        producer.close();
+        consumerTwo.close();
+
+        pulsar.getConfig().setAllowAutoUpdateSchema(false);
+
+        producer = producerThreeBuilder.create();
+        consumerTwo = comsumerBuilder.subscribe();
+
+        producer.send(new Schemas.PersonTwo(2, "Lucy"));
+        message = consumerTwo.receive();
+
+        personTwo = message.getValue();
+        consumerTwo.acknowledge(message);
+
+        assertEquals(personTwo.getId(), 2);
+        assertEquals(personTwo.getName(), "Lucy");
+
+        consumerTwo.close();
+        producer.close();
+    }
+
     @Test(dataProvider =  "AllCheckSchemaCompatibilityStrategy")
     public void testIsAutoUpdateSchema(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
         final String tenant = PUBLIC_TENANT;

From 4089da05ac217b8a4ea46bf0a58fdfbe7ce17cbb Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Sun, 14 Nov 2021 09:44:16 +0800
Subject: [PATCH 2/6] update isAllowAutoUpdateSchema to allowAutoUpdateSchema

---
 .../java/org/apache/pulsar/broker/ServiceConfiguration.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

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 1efca41d9c6c6..1c89c7518a3b1 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
@@ -579,7 +579,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace "
             + "policy will be ignored. This is enabled by default."
     )
-    private boolean isAllowAutoUpdateSchema = true;
+    private boolean allowAutoUpdateSchema = true;
 
     @FieldContext(
         category = CATEGORY_SERVER,

From 80c25de9785bf0e3f4548440166bfbf188427cc3 Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Sun, 14 Nov 2021 16:12:06 +0800
Subject: [PATCH 3/6] add config in conf/broker.conf and conf/standalone.conf

---
 conf/broker.conf     | 4 ++++
 conf/standalone.conf | 4 ++++
 2 files changed, 8 insertions(+)

diff --git a/conf/broker.conf b/conf/broker.conf
index 13e955bdee53d..7917a4daeef66 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -270,6 +270,10 @@ brokerMaxConnections=0
 # The maximum number of connections per IP. If it exceeds, new connections are rejected.
 brokerMaxConnectionsPerIp=0
 
+# Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
+# policy will be ignored. This is enabled by default.
+allowAutoUpdateSchema=true
+
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 87d9e058f465c..dab5c5c833674 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -176,6 +176,10 @@ defaultNumberOfNamespaceBundles=4
 # Using a value of 0, is disabling maxTopicsPerNamespace-limit check.
 maxTopicsPerNamespace=0
 
+# Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
+# policy will be ignored. This is enabled by default.
+allowAutoUpdateSchema=true
+
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
 

From 68005531069eaacf656351cf4387c29a0a4eea2b Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Mon, 15 Nov 2021 15:01:53 +0800
Subject: [PATCH 4/6] add config doc in
 site2/website-next/docs/reference-configuration.md

---
 site2/website-next/docs/reference-configuration.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/site2/website-next/docs/reference-configuration.md b/site2/website-next/docs/reference-configuration.md
index 1908149053ba1..f6a5dddb64285 100644
--- a/site2/website-next/docs/reference-configuration.md
+++ b/site2/website-next/docs/reference-configuration.md
@@ -349,6 +349,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater
 | managedLedgerInfoCompressionType | Compression type of managed ledger information. <br /><br />Available options are `NONE`, `LZ4`, `ZLIB`, `ZSTD`, and `SNAPPY`). <br /><br />If this value is `NONE` or invalid, the `managedLedgerInfo` is not compressed. <br /><br />**Note** that after enabling this configuration, if you want to degrade a broker, you need to change the value to `NONE` and make sure all ledger metadata is saved without compression. | None |
 | additionalServlets | Additional servlet name. <br /><br />If you have multiple additional servlets, separate them by commas. <br /><br />For example, additionalServlet_1, additionalServlet_2 | N/A |
 | additionalServletDirectory | Location of broker additional servlet NAR directory | ./brokerAdditionalServlet |
+| allowAutoUpdateSchema | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
 
 ## Client
 
@@ -480,7 +481,6 @@ You can set the log level and configuration in the  [log4j2.yaml](https://github
 | dispatchThrottlingRatePerTopicInMsg | Default messages (per second) dispatch throttling-limit for every topic. When the value is set to 0, default message dispatch throttling-limit is disabled. |0 |
 | dispatchThrottlingRatePerTopicInByte | Default byte (per second) dispatch throttling-limit for every topic. When the value is set to 0, default byte dispatch throttling-limit is disabled. | 0|
 | dispatchThrottlingOnBatchMessageEnabled |Apply dispatch rate limiting on batch message instead individual messages with in batch message. (Default is disabled). | false|
-
 | dispatchThrottlingRateRelativeToPublishRate | Enable dispatch rate-limiting relative to publish rate. | false |
 |dispatchThrottlingRatePerSubscriptionInMsg|The defaulted number of message dispatching throttling-limit for a subscription. The value of 0 disables message dispatch-throttling.|0|
 |dispatchThrottlingRatePerSubscriptionInByte|The default number of message-bytes dispatching throttling-limit for a subscription. The value of 0 disables message-byte dispatch-throttling.|0|
@@ -650,6 +650,7 @@ You can set the log level and configuration in the  [log4j2.yaml](https://github
 |haProxyProtocolEnabled | Enable or disable the [HAProxy](http://www.haproxy.org/) protocol. |false|
 |bookieId | If you want to custom a bookie ID or use a dynamic network address for a bookie, you can set the `bookieId`. <br /><br />Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`).<br /><br /> The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots. <br /><br />For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|/|
 | maxTopicsPerNamespace | The maximum number of persistent topics that can be created in the namespace. When the number of topics reaches this threshold, the broker rejects the request of creating a new topic, including the auto-created topics by the producer or consumer, until the number of connected consumers decreases. The default value 0 disables the check. | 0 |
+| allowAutoUpdateSchema | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
 
 ## WebSocket
 

From 3eddf8ac256f8762c12ff8d7cf0a7a0ed0a2ccd2 Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Thu, 18 Nov 2021 16:47:08 +0800
Subject: [PATCH 5/6] update by comment

---
 conf/broker.conf                                      |  2 +-
 conf/standalone.conf                                  |  2 +-
 .../apache/pulsar/broker/ServiceConfiguration.java    |  2 +-
 .../apache/pulsar/broker/service/AbstractTopic.java   | 11 +++++++++--
 .../compatibility/SchemaCompatibilityCheckTest.java   |  6 +++---
 .../apache/pulsar/common/policies/data/Policies.java  |  2 +-
 site2/website-next/docs/reference-configuration.md    |  4 ++--
 7 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/conf/broker.conf b/conf/broker.conf
index 7917a4daeef66..e8d1e2c8ab044 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -272,7 +272,7 @@ brokerMaxConnectionsPerIp=0
 
 # Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
 # policy will be ignored. This is enabled by default.
-allowAutoUpdateSchema=true
+isAllowAutoUpdateSchemaEnabled=true
 
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
diff --git a/conf/standalone.conf b/conf/standalone.conf
index dab5c5c833674..6f03f18a6795d 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -178,7 +178,7 @@ maxTopicsPerNamespace=0
 
 # Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
 # policy will be ignored. This is enabled by default.
-allowAutoUpdateSchema=true
+isAllowAutoUpdateSchemaEnabled=true
 
 # Enable check for minimum allowed client library version
 clientLibraryVersionCheckEnabled=false
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 1c89c7518a3b1..b73e6c839d14b 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
@@ -579,7 +579,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace "
             + "policy will be ignored. This is enabled by default."
     )
-    private boolean allowAutoUpdateSchema = true;
+    private boolean isAllowAutoUpdateSchemaEnabled = true;
 
     @FieldContext(
         category = CATEGORY_SERVER,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 0511c05bf255d..73885678bf9ee 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -98,7 +98,7 @@ public abstract class AbstractTopic implements Topic {
     @Getter
     protected volatile SchemaCompatibilityStrategy schemaCompatibilityStrategy =
             SchemaCompatibilityStrategy.FULL;
-    protected volatile boolean isAllowAutoUpdateSchema = true;
+    protected volatile Boolean isAllowAutoUpdateSchema;
     // schema validation enforced flag
     protected volatile boolean schemaValidationEnforced = false;
 
@@ -334,7 +334,7 @@ public CompletableFuture<SchemaVersion> addSchema(SchemaData schema) {
         String id = TopicName.get(base).getSchemaName();
         SchemaRegistryService schemaRegistryService = brokerService.pulsar().getSchemaRegistryService();
 
-        if (brokerService.pulsar().getConfig().isAllowAutoUpdateSchema() && isAllowAutoUpdateSchema) {
+        if (allowAutoUpdateSchema()) {
             return schemaRegistryService.putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy);
         } else {
             return schemaRegistryService.trimDeletedSchemaAndGetList(id).thenCompose(schemaAndMetadataList ->
@@ -350,6 +350,13 @@ public CompletableFuture<SchemaVersion> addSchema(SchemaData schema) {
         }
     }
 
+    private boolean allowAutoUpdateSchema() {
+        if (isAllowAutoUpdateSchema == null) {
+            return brokerService.pulsar().getConfig().isAllowAutoUpdateSchemaEnabled();
+        }
+        return isAllowAutoUpdateSchema;
+    }
+
     @Override
     public CompletableFuture<SchemaVersion> deleteSchema() {
         String base = TopicName.get(getName()).getPartitionedTopicName();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
index b169723dc4463..80168b9ae4c9f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
@@ -246,7 +246,7 @@ public void testBrokerAllowAutoUpdateSchemaDisabled(SchemaCompatibilityStrategy
         admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
 
 
-        pulsar.getConfig().setAllowAutoUpdateSchema(false);
+        pulsar.getConfig().setAllowAutoUpdateSchemaEnabled(false);
         ProducerBuilder<Schemas.PersonTwo> producerThreeBuilder = pulsarClient
                 .newProducer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
                                 (false).withSupportSchemaVersioning(true).
@@ -258,7 +258,7 @@ public void testBrokerAllowAutoUpdateSchemaDisabled(SchemaCompatibilityStrategy
             Assert.assertTrue(e.getMessage().contains("Schema not found and schema auto updating is disabled."));
         }
 
-        pulsar.getConfig().setAllowAutoUpdateSchema(true);
+        pulsar.getConfig().setAllowAutoUpdateSchemaEnabled(true);
         ConsumerBuilder<Schemas.PersonTwo> comsumerBuilder = pulsarClient.newConsumer(Schema.AVRO(
                         SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
                                         (false).withSupportSchemaVersioning(true).
@@ -281,7 +281,7 @@ public void testBrokerAllowAutoUpdateSchemaDisabled(SchemaCompatibilityStrategy
         producer.close();
         consumerTwo.close();
 
-        pulsar.getConfig().setAllowAutoUpdateSchema(false);
+        pulsar.getConfig().setAllowAutoUpdateSchemaEnabled(false);
 
         producer = producerThreeBuilder.create();
         consumerTwo = comsumerBuilder.subscribe();
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
index 7f56bc8e50735..4d29a5fdee669 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
@@ -107,7 +107,7 @@ public class Policies {
     public SchemaCompatibilityStrategy schema_compatibility_strategy = SchemaCompatibilityStrategy.UNDEFINED;
 
     @SuppressWarnings("checkstyle:MemberName")
-    public boolean is_allow_auto_update_schema = true;
+    public Boolean is_allow_auto_update_schema = null;
 
     @SuppressWarnings("checkstyle:MemberName")
     public boolean schema_validation_enforced = false;
diff --git a/site2/website-next/docs/reference-configuration.md b/site2/website-next/docs/reference-configuration.md
index f6a5dddb64285..aed529e1f0682 100644
--- a/site2/website-next/docs/reference-configuration.md
+++ b/site2/website-next/docs/reference-configuration.md
@@ -349,7 +349,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater
 | managedLedgerInfoCompressionType | Compression type of managed ledger information. <br /><br />Available options are `NONE`, `LZ4`, `ZLIB`, `ZSTD`, and `SNAPPY`). <br /><br />If this value is `NONE` or invalid, the `managedLedgerInfo` is not compressed. <br /><br />**Note** that after enabling this configuration, if you want to degrade a broker, you need to change the value to `NONE` and make sure all ledger metadata is saved without compression. | None |
 | additionalServlets | Additional servlet name. <br /><br />If you have multiple additional servlets, separate them by commas. <br /><br />For example, additionalServlet_1, additionalServlet_2 | N/A |
 | additionalServletDirectory | Location of broker additional servlet NAR directory | ./brokerAdditionalServlet |
-| allowAutoUpdateSchema | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
+| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
 
 ## Client
 
@@ -650,7 +650,7 @@ You can set the log level and configuration in the  [log4j2.yaml](https://github
 |haProxyProtocolEnabled | Enable or disable the [HAProxy](http://www.haproxy.org/) protocol. |false|
 |bookieId | If you want to custom a bookie ID or use a dynamic network address for a bookie, you can set the `bookieId`. <br /><br />Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`).<br /><br /> The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots. <br /><br />For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|/|
 | maxTopicsPerNamespace | The maximum number of persistent topics that can be created in the namespace. When the number of topics reaches this threshold, the broker rejects the request of creating a new topic, including the auto-created topics by the producer or consumer, until the number of connected consumers decreases. The default value 0 disables the check. | 0 |
-| allowAutoUpdateSchema | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
+| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
 
 ## WebSocket
 

From 3619de5852df9e76857b6b577ba81ad47320e44e Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Thu, 18 Nov 2021 17:15:17 +0800
Subject: [PATCH 6/6] update doc

---
 conf/broker.conf                                              | 4 ++--
 conf/standalone.conf                                          | 4 ++--
 .../java/org/apache/pulsar/broker/ServiceConfiguration.java   | 4 ++--
 site2/website-next/docs/reference-configuration.md            | 4 ++--
 4 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/conf/broker.conf b/conf/broker.conf
index e8d1e2c8ab044..700f9a5adc9b1 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -270,8 +270,8 @@ brokerMaxConnections=0
 # The maximum number of connections per IP. If it exceeds, new connections are rejected.
 brokerMaxConnectionsPerIp=0
 
-# Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
-# policy will be ignored. This is enabled by default.
+# Allow schema to be auto updated at broker level. User can override this by
+# 'is_allow_auto_update_schema' of namespace policy.
 isAllowAutoUpdateSchemaEnabled=true
 
 # Enable check for minimum allowed client library version
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 6f03f18a6795d..906280c180253 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -176,8 +176,8 @@ defaultNumberOfNamespaceBundles=4
 # Using a value of 0, is disabling maxTopicsPerNamespace-limit check.
 maxTopicsPerNamespace=0
 
-# Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace
-# policy will be ignored. This is enabled by default.
+# Allow schema to be auto updated at broker level. User can override this by
+# 'is_allow_auto_update_schema' of namespace policy.
 isAllowAutoUpdateSchemaEnabled=true
 
 # Enable check for minimum allowed client library version
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 b73e6c839d14b..9ca086a4c7d67 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
@@ -576,8 +576,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(
         category = CATEGORY_POLICIES,
         dynamic = true,
-        doc = "Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace "
-            + "policy will be ignored. This is enabled by default."
+        doc = "Allow schema to be auto updated at broker level. User can override this by 'is_allow_auto_update_schema'"
+            + " of namespace policy. This is enabled by default."
     )
     private boolean isAllowAutoUpdateSchemaEnabled = true;
 
diff --git a/site2/website-next/docs/reference-configuration.md b/site2/website-next/docs/reference-configuration.md
index aed529e1f0682..3199155d3aaad 100644
--- a/site2/website-next/docs/reference-configuration.md
+++ b/site2/website-next/docs/reference-configuration.md
@@ -349,7 +349,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater
 | managedLedgerInfoCompressionType | Compression type of managed ledger information. <br /><br />Available options are `NONE`, `LZ4`, `ZLIB`, `ZSTD`, and `SNAPPY`). <br /><br />If this value is `NONE` or invalid, the `managedLedgerInfo` is not compressed. <br /><br />**Note** that after enabling this configuration, if you want to degrade a broker, you need to change the value to `NONE` and make sure all ledger metadata is saved without compression. | None |
 | additionalServlets | Additional servlet name. <br /><br />If you have multiple additional servlets, separate them by commas. <br /><br />For example, additionalServlet_1, additionalServlet_2 | N/A |
 | additionalServletDirectory | Location of broker additional servlet NAR directory | ./brokerAdditionalServlet |
-| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
+| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated at broker level. User can override this by 'is_allow_auto_update_schema' of namespace policy. |true|
 
 ## Client
 
@@ -650,7 +650,7 @@ You can set the log level and configuration in the  [log4j2.yaml](https://github
 |haProxyProtocolEnabled | Enable or disable the [HAProxy](http://www.haproxy.org/) protocol. |false|
 |bookieId | If you want to custom a bookie ID or use a dynamic network address for a bookie, you can set the `bookieId`. <br /><br />Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`).<br /><br /> The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots. <br /><br />For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|/|
 | maxTopicsPerNamespace | The maximum number of persistent topics that can be created in the namespace. When the number of topics reaches this threshold, the broker rejects the request of creating a new topic, including the auto-created topics by the producer or consumer, until the number of connected consumers decreases. The default value 0 disables the check. | 0 |
-| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated. If this is disabled, 'is_allow_auto_update_schema' in namespace policy will be ignored. This is enabled by default.|true|
+| isAllowAutoUpdateSchemaEnabled | Allow schema to be auto updated at broker level. User can override this by 'is_allow_auto_update_schema' of namespace policy. |true|
 
 ## WebSocket