made changes as per code review comments

Author: aparajita-singh-flipkart

conf/broker.conf

@@ -555,8 +555,7 @@ maxNumPartitionsPerPartitionedTopic=0
 # There are two policies to apply when broker metadata session expires: session expired happens, "shutdown" or "reconnect".
 # With "shutdown", the broker will be restarted.
 # With "reconnect", the broker will keep serving the topics, while attempting to recreate a new session.
-# Node: the "reconnect" policy is an experiment feature
-zookeeperSessionExpiredPolicy=shutdown
+zookeeperSessionExpiredPolicy=reconnect
 
 # Enable or disable system topic
 systemTopicEnabled=false

deployment/terraform-ansible/templates/broker.conf

@@ -424,8 +424,7 @@ maxNumPartitionsPerPartitionedTopic=0
 # There are two policies when zookeeper session expired happens, "shutdown" and "reconnect".
 # If uses "shutdown" policy, shutdown the broker when zookeeper session expired happens.
 # If uses "reconnect" policy, try to reconnect to zookeeper server and re-register metadata to zookeeper.
-# Node: the "reconnect" policy is an experiment feature
-zookeeperSessionExpiredPolicy=shutdown
+zookeeperSessionExpiredPolicy=reconnect
 
 # Enable or disable system topic
 systemTopicEnabled=false

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -1176,7 +1176,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
         + " With \"shutdown\", the broker will be restarted.\n\n"
         + " With \"reconnect\", the broker will keep serving the topics, while attempting to recreate a new session."
     )
-    private MetadataSessionExpiredPolicy zookeeperSessionExpiredPolicy = MetadataSessionExpiredPolicy.shutdown;
+    private MetadataSessionExpiredPolicy zookeeperSessionExpiredPolicy = MetadataSessionExpiredPolicy.reconnect;
 
     @FieldContext(
         category = CATEGORY_SERVER,

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -674,7 +674,8 @@ public void start() throws PulsarServerException {
             this.startNamespaceService();
 
             schemaStorage = createAndStartSchemaStorage();
-            schemaRegistryService = SchemaRegistryService.create(schemaStorage, config.getSchemaRegistryClassName());
+            setSchemaRegistryName(schemaStorage);
+            schemaRegistryService = SchemaRegistryService.create(config.getSchemaRegistryClassName());
             schemaRegistryService.initialize(config, schemaStorage);
 
             this.defaultOffloader = createManagedLedgerOffloader(
@@ -1266,6 +1267,12 @@ private SchemaStorage createAndStartSchemaStorage() throws Exception {
         return schemaStorage;
     }
 
+    private void setSchemaRegistryName(SchemaStorage schemaStorage) {
+        if (schemaStorage == null) {
+            this.config.setSchemaRegistryClassName("org.apache.pulsar.broker.service.schema.DefaultSchemaRegistryService");
+        }
+    }
+
     public ScheduledExecutorService getExecutor() {
         return executor;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/DefaultSchemaRegistryService.java

@@ -108,9 +108,7 @@ public SchemaVersion versionFromBytes(byte[] version) {
 
     @Override
     public void initialize(ServiceConfiguration configuration, SchemaStorage schemaStorage)
-          throws PulsarServerException {
-
-    }
+          throws PulsarServerException {}
 
     @Override
     public void close() throws Exception {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryService.java

@@ -29,17 +29,14 @@ public interface SchemaRegistryService extends SchemaRegistry {
     Logger LOG = LoggerFactory.getLogger(SchemaRegistryService.class);
     long NO_SCHEMA_VERSION = -1L;
 
-    static SchemaRegistryService create(SchemaStorage schemaStorage, String schemaRegistryClassName) {
-        if (schemaStorage != null) {
-            try {
-                SchemaRegistryService schemaRegistryService = (SchemaRegistryService) Class
-                      .forName(schemaRegistryClassName).getDeclaredConstructor()
-                      .newInstance();
+    static SchemaRegistryService create(String schemaRegistryClassName) {
+        try {
+            SchemaRegistryService schemaRegistryService = (SchemaRegistryService) Class.forName(schemaRegistryClassName)
+                  .getDeclaredConstructor().newInstance();
 
-                return SchemaRegistryServiceWithSchemaDataValidator.of(schemaRegistryService);
-            } catch (Exception e) {
-                LOG.warn("Unable to create schema registry storage, defaulting to empty storage", e);
-            }
+            return SchemaRegistryServiceWithSchemaDataValidator.of(schemaRegistryService);
+        } catch (Exception e) {
+            LOG.warn("Unable to create schema registry storage, defaulting to empty storage", e);
         }
         return new DefaultSchemaRegistryService();
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java

@@ -25,7 +25,6 @@
 import static org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE;
 import static org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy.FORWARD_TRANSITIVE;
 import static org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy.FULL_TRANSITIVE;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
@@ -53,7 +52,6 @@
 import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
 import org.apache.pulsar.broker.service.schema.exceptions.SchemaException;
 import org.apache.pulsar.broker.service.schema.proto.SchemaRegistryFormat;
-import org.apache.pulsar.broker.service.schema.validator.SchemaRegistryServiceWithSchemaDataValidator;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.protocol.schema.SchemaHash;
@@ -218,7 +216,6 @@ public SchemaVersion versionFromBytes(byte[] version) {
     public void initialize(ServiceConfiguration configuration, SchemaStorage schemaStorage) throws PulsarServerException {
         try {
             Map<SchemaType, SchemaCompatibilityCheck> checkers = getCheckers(configuration.getSchemaRegistryCompatibilityCheckers());
-            checkers.put(SchemaType.KEY_VALUE, new KeyValueSchemaCompatibilityCheck(checkers));
             this.schemaStorage = schemaStorage;
             this.compatibilityChecks = checkers;
             this.clock = Clock.systemUTC();
@@ -237,6 +234,7 @@ private Map<SchemaType, SchemaCompatibilityCheck> getCheckers(Set<String> checke
                   .getDeclaredConstructor().newInstance();
             checkers.put(instance.getSchemaType(), instance);
         }
+        checkers.put(SchemaType.KEY_VALUE, new KeyValueSchemaCompatibilityCheck(checkers));
         return checkers;
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceTest.java

@@ -45,18 +45,18 @@ public void setup() {
    @Test
    void createCustomRegistry() {
       SchemaRegistryService actual = SchemaRegistryService
-            .create(mockSchemaStorage, "org.apache.pulsar.broker.service.schema.TestSchemaRegistryService");
+            .create("org.apache.pulsar.broker.service.schema.TestSchemaRegistryService");
       Assert.assertSame(actual.getClass(), SchemaRegistryServiceWithSchemaDataValidator.class);
    }
 
    @Test
    void createEmptyRegistry() {
-      SchemaRegistryService actual = SchemaRegistryService.create(null, null);
+      SchemaRegistryService actual = SchemaRegistryService.create(null);
       Assert.assertSame(actual.getClass(), DefaultSchemaRegistryService.class);
    }
 
 }
 
 class TestSchemaRegistryService extends DefaultSchemaRegistryService {
-   public TestSchemaRegistryService(SchemaStorage schemaStorage, Map checkers) {}
+   public TestSchemaRegistryService() {}
 }

site2/website-next/versioned_docs/version-2.7.3/reference-configuration.md

@@ -235,7 +235,6 @@ Pulsar brokers are responsible for handling incoming messages from producers, di
 |athenzDomainNames| Supported Athenz provider domain names(comma separated) for authentication  ||
 |exposePreciseBacklogInPrometheus| Enable expose the precise backlog stats, set false to use published counter and consumed counter to calculate, this would be more efficient but may be inaccurate. |false|
 |schemaRegistryStorageClassName|The schema storage implementation used by this broker.|org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageFactory|
-|schemaRegistryClassName|Specify the schema registry to be used in Pulsar. If it is not set, the default schema registry is used.|org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl|
 |isSchemaValidationEnforced|Enforce schema validation on following cases: if a producer without a schema attempts to produce to a topic with schema, the producer will be failed to connect. PLEASE be carefully on using this, since non-java clients don't support schema. If this setting is enabled, then non-java clients fail to produce.|false|
 |offloadersDirectory|The directory for all the offloader implementations.|./offloaders|
 |bookkeeperMetadataServiceUri| Metadata service uri that bookkeeper is used for loading corresponding metadata driver and resolving its metadata service location. This value can be fetched using `bookkeeper shell whatisinstanceid` command in BookKeeper cluster. For example: zk+hierarchical://localhost:2181/ledgers. The metadata service uri list can also be semicolon separated values like below: zk+hierarchical://zk1:2181;zk2:2181;zk3:2181/ledgers ||

site2/website/versioned_docs/version-2.9.1/reference-configuration.md

@@ -241,6 +241,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater
 |athenzDomainNames| Supported Athenz provider domain names(comma separated) for authentication  ||
 |exposePreciseBacklogInPrometheus| Enable expose the precise backlog stats, set false to use published counter and consumed counter to calculate, this would be more efficient but may be inaccurate. |false|
 |schemaRegistryStorageClassName|The schema storage implementation used by this broker.|org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageFactory|
+|schemaRegistryClassName|Specify the schema registry to be used in Pulsar. If it is not set, the default schema registry is used.|org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl|
 |isSchemaValidationEnforced|Enforce schema validation on following cases: if a producer without a schema attempts to produce to a topic with schema, the producer will be failed to connect. PLEASE be carefully on using this, since non-java clients don't support schema. If this setting is enabled, then non-java clients fail to produce.|false|
 | topicFencingTimeoutSeconds | If a topic remains fenced for a certain time period (in seconds), it is closed forcefully. If set to 0 or a negative number, the fenced topic is not closed. | 0 |
 |offloadersDirectory|The directory for all the offloader implementations.|./offloaders|