PIP-144: Making SchemaRegistry implementation configurable #14102

aparajita89 · 2022-02-03T08:18:11Z

Motivation

SchemaRegistryService is hardcoded to use SchemaRegistryServiceImpl in org.apache.pulsar.broker.service.schema.SchemaRegistryService#create

Modifications

broker.conf has a new config called schemaRegistryClassName which will have the name of the class to be used. the class is instantiated in SchemaRegistryService interface's static create method via reflection.

Verifying this change

Make sure that the change passes the CI checks.

This change is already covered by existing tests which create a producer or consumer successfully.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): no
The public API: no
The schema: yes
The default values of configurations: yes
The wire protocol: no
The rest endpoints: no
The admin cli options: no
Anything that affects deployment: no

Documentation

Check the box below or label this PR directly (if you have committer privilege).

Need to update docs?

doc-required
this PR adds a new config which needs to be explained in the docs

codelipenghui · 2022-02-09T03:41:23Z

conf/standalone.conf

@@ -180,6 +180,8 @@ maxTopicsPerNamespace=0
 # 'is_allow_auto_update_schema' of namespace policy.
 isAllowAutoUpdateSchemaEnabled=true

+schemaRegistryClass=


It's better to have a description, by default pulsar will use the internal schema registry which is based on the bookkeeper.

@codelipenghui i have added it now

codelipenghui · 2022-02-09T03:41:40Z

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

@@ -42,19 +44,29 @@
        return checkers;
    }

-    static SchemaRegistryService create(SchemaStorage schemaStorage, Set<String> schemaRegistryCompatibilityCheckers) {
+    static SchemaRegistryService create(String schemaRegistryName, SchemaStorage schemaStorage,


Suggested change

static SchemaRegistryService create(String schemaRegistryName, SchemaStorage schemaStorage,

static SchemaRegistryService create(String schemaRegistryClassName, SchemaStorage schemaStorage,

this is updated

codelipenghui · 2022-02-09T03:42:45Z

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

+                return SchemaRegistryServiceWithSchemaDataValidator
+                      .of(new SchemaRegistryServiceImpl(schemaStorage, checkers));


Correct the code format there.

codelipenghui · 2022-02-09T03:43:10Z

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

+                return SchemaRegistryServiceWithSchemaDataValidator
+                      .of(new SchemaRegistryServiceImpl(schemaStorage, checkers));
+                } else {
+                    return (SchemaRegistryService) Class.forName(schemaRegistryName)


Better to have a unit test to make sure the configuration works well.

the default case is already tested in the existing test cases. i'll add some more test cases to explicitly test this part.

have updated this logic. we are using a default value in the config and continuing to use the schema validator wrapper class.
the reasoning was that any custom schema registry implementation would need to comply with existing usages of SchemaRegistryService interface, including pulsar-admin apis and the client connection flows. the SchemaRegistryServiceWithSchemaDataValidator wrapper class ensures that any input schemas are validated via avro/json/protobuf-native parser before the compatibility checks and schema storage interactions are applied.

aparajita89 · 2022-02-09T06:52:44Z

comments from yesterday's review by @merlimat and ankur:

schema storage might not be used in custom schema registry, it need not be passed in the constructor
custom schema registry will require additional configs which will be present in ServiceConfiguration but it's not being passed to the constructor right now
evaluate the existing approach once more keeping in mind the above points

Huanli-Meng · 2022-02-10T06:00:12Z

@aparajita89 do we also need to update the broker.config file? Thanks
@Anonymitaet could you or Jun help follow up adding / updating docs about the PR? Thanks.

aparajita89 · 2022-02-10T09:49:55Z

@Huanli-Meng this is an optional config. if no value is provided in the broker.conf file then the existing flow would go through. imo, we can add this config to the documentation but there isn't a need to update the config files unless someone really requires it.

aparajita89 · 2022-02-10T12:13:50Z

@Anonymitaet could you let me know what changes i need to make for the documentation?

Anonymitaet · 2022-02-10T12:30:49Z

@aparajita89 how about adding the parameter and its description here https://pulsar.apache.org/docs/en/next/reference-configuration/#standalone?

aparajita89 · 2022-02-11T04:04:30Z

@Anonymitaet thanks, i have added it in reference-configuration.md

Huanli-Meng

@aparajita89 only one minor comment. PTAL

Huanli-Meng · 2022-02-11T04:12:30Z

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

@@ -235,6 +235,7 @@ 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|Override the schema registry used by pulsar with a custom implementation. If this config is not provided, the default schema registry will be used.|org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl|


Could we change the description of the parameter as below ?
Specify the schema registry to be used in Pulsar. If it is not set, the default schema registry is used.

sure, i have updated it

Huanli-Meng

Approve from technical writer's points of view. Let's wait for technical approval.

315157973 · 2022-02-15T06:11:43Z

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

            } catch (Exception e) {
                LOG.warn("Unable to create schema registry storage, defaulting to empty storage", e);
            }
        }
        return new DefaultSchemaRegistryService();
    }

+    void initialize(ServiceConfiguration configuration);


Why add this interface? It doesn't seem necessary

this would be needed in custom implementations of the schema registry where additional configs are required which are specific to the custom implementation. for example, we require a custom schema registry which needs to make HTTP calls to another service. configs specific to the schema registry, like the service endpoint, needs to be passed via ServiceConfiguration and set in the schema registry using this initialize method. this is the same convention being followed for other configs such as authorization provider as well.

anvinjain · 2022-02-15T17:54:04Z

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

@@ -235,6 +235,7 @@ 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|


doc changes should be in recent pulsar versions as well right?

moved this to the latest doc as the change will be in newer versions

eolivelli

Hello @aparajita89

Thanks for your initiative.
Can you please start a PIP and a discussion on [email protected] for this feature?

aparajita89 · 2022-02-17T12:18:16Z

Hello @aparajita89

Thanks for your initiative. Can you please start a PIP and a discussion on [email protected] for this feature?

sure, will do that

aparajita89 · 2022-02-21T07:29:44Z

@eolivelli the PIP has been created, please take a look: #14395

aparajita89 · 2022-04-19T05:37:26Z

@eolivelli i have added the integration test

congbobo184 · 2022-04-20T02:26:11Z

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

@@ -706,8 +707,9 @@ public void start() throws PulsarServerException {
            this.startNamespaceService();

            schemaStorage = createAndStartSchemaStorage();
-            schemaRegistryService = SchemaRegistryService.create(
-                    schemaStorage, config.getSchemaRegistryCompatibilityCheckers());
+            ensureSchemaRegistryName(schemaStorage);


createAndStartSchemaStorage will create schemaStorage, if don't set the schemaRegistryStorageClassName, it will throw NPE, so this method seem not useful or need to add a default SchemaRegistryStorageClassName

@congbobo184 the default schema registry name is being set in ServiceConfiguration so NPE would not happen here

I mean why we need check schemaStorage == null in ensureSchemaRegistryName

this was part of the existing logic. if schema storage is null then the SchemaRegistryServiceDisabled needs to be used. otherwise, SchemaRegistryServiceImpl needs to be used by default.
below is the expected behavior for each of the combinations for configs set in broker's conf file.

if schemaRegistryClassName == null and schemaRegistryStorageClassName == null then set schemaStorage to null and schemaRegistry to SchemaRegistryServiceDisabled

if schemaRegistryClassName != null and schemaRegistryStorageClassName == null then set schemaStorage to null and schemaRegistry to SchemaRegistryServiceDisabled

if schemaRegistryClassName == null and schemaRegistryStorageClassName != null then set schemaStorage to the configured object and schemaRegistry to SchemaRegistryServiceImpl

if schemaRegistryClassName != null and schemaRegistryStorageClassName != null then set schemaStorage to the configured object and schemaRegistry to the configured object

in the older code, this logic was implemented in SchemaRegistryService.create(...)

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

Line 1303 in caf7648

SchemaStorage schemaStorage = (SchemaStorage) createMethod.invoke(factoryInstance, this);

in this code, createAndStartSchemaStorage will not create a schemaStorage is null. so why we should check ensureSchemaRegistryName and set default?

you are right, schemaStorage can never be null. i had only kept this null check to retain the existing logic. i have removed it now.

codelipenghui · 2022-04-21T09:41:42Z

@eolivelli Please help review this PR again.

codelipenghui · 2022-04-25T01:07:42Z

ping @eolivelli Please help review this PR.

eolivelli · 2022-05-18T11:13:16Z

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

+            result.complete(new SchemaAndMetadata(schemaId, schemaData,
+                    storedSchemaCompletableFuture.get().get(version).get().version));
+        } catch (InterruptedException e) {
+            e.printStackTrace();


please use logger

eolivelli · 2022-05-18T11:13:23Z

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

+            result.complete(new SchemaAndMetadata(schemaId, schemaData,
+                    storedSchemaCompletableFuture.get().get(versionInt).get().version));
+        } catch (InterruptedException e) {
+            e.printStackTrace();


please use logger

eolivelli · 2022-05-18T11:13:46Z

pulsar-common/src/main/java/org/apache/pulsar/common/naming/Constants.java

@@ -24,6 +24,11 @@
 public class Constants {

    public static final String GLOBAL_CLUSTER = "global";
+    public static final String SCHEMA_REGISTRY_CLASS_NAME;


nit: please declare the value here: "org.apache.pulsar.broker.service.schema.SchemaRegistryServiceDisabled"

eolivelli · 2022-05-18T11:14:26Z

...integration/src/test/java/org/apache/pulsar/tests/integration/schema/SchemaRegistryTest.java

+@Slf4j
+public class SchemaRegistryTest extends TopicMessagingBase {
+    protected String methodName;
+    private String SCHEMA_REGISTRY="org.apache.pulsar.broker.service.schema.MockSchemaRegistry";


nit: static ? also please add a white space around the '=' char

github-actions · 2022-06-18T02:15:38Z

The pr had no activity for 30 days, mark with Stale label.

github-actions bot assigned aparajita89 Feb 3, 2022

github-actions bot added the doc-required Your PR changes impact docs and you will update later. label Feb 3, 2022

sijie requested review from merlimat, congbobo184, codelipenghui, hangc0276 and Huanli-Meng February 9, 2022 02:38

codelipenghui reviewed Feb 9, 2022

View reviewed changes

Huanli-Meng reviewed Feb 11, 2022

View reviewed changes

Huanli-Meng previously approved these changes Feb 11, 2022

View reviewed changes

Anonymitaet added doc Your PR contains doc changes, no matter whether the changes are in markdown or code files. and removed doc-required Your PR changes impact docs and you will update later. labels Feb 14, 2022

codelipenghui added this to the 2.11.0 milestone Feb 15, 2022

codelipenghui previously approved these changes Feb 15, 2022

View reviewed changes

315157973 reviewed Feb 15, 2022

View reviewed changes

315157973 previously approved these changes Feb 15, 2022

View reviewed changes

anvinjain reviewed Feb 15, 2022

View reviewed changes

aparajita89 dismissed stale reviews from 315157973, codelipenghui, and Huanli-Meng via 020ac8a February 17, 2022 11:13

eolivelli requested changes Feb 17, 2022

View reviewed changes

aparajita89 mentioned this pull request Feb 21, 2022

PIP-144: Making SchemaRegistry implementation configurable #14395

Open

aparajita-singh-flipkart added 2 commits April 18, 2022 16:51

resolving conflicts

6b001ec

added tests

a55473e

congbobo184 approved these changes Apr 20, 2022

View reviewed changes

aparajita-singh-flipkart added 4 commits April 20, 2022 12:10

renamed default schema registry

a09138b

Merge branch 'master' of github.com:apache/pulsar into schema-1

d5c1cb1

fixed checkstyle

909c5cd

code review comments

8d508eb

codelipenghui approved these changes Apr 21, 2022

View reviewed changes

aparajita-singh-flipkart added 2 commits April 22, 2022 11:20

Merge branch 'master' of github.com:apache/pulsar into schema-1

5fed20c

checkstyle fix and back to master

6260106

aparajita-singh-flipkart added 2 commits April 25, 2022 12:58

license check fixes

f99a43f

Merge branch 'master' of github.com:apache/pulsar into schema-1

e05545c

eolivelli requested changes May 18, 2022

View reviewed changes

github-actions bot added the Stale label Jun 18, 2022

codelipenghui modified the milestones: 2.11.0, 2.12.0 Jul 26, 2022

RobertIndie modified the milestones: 3.0.0, 3.1.0 Apr 10, 2023

Technoboy- modified the milestones: 3.1.0, 3.2.0 Jul 31, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

lhotari modified the milestones: 4.0.0, 4.1.0 Oct 14, 2024

	static SchemaRegistryService create(String schemaRegistryName, SchemaStorage schemaStorage,
	static SchemaRegistryService create(String schemaRegistryClassName, SchemaStorage schemaStorage,

		return SchemaRegistryServiceWithSchemaDataValidator
		.of(new SchemaRegistryServiceImpl(schemaStorage, checkers));

PIP-144: Making SchemaRegistry implementation configurable #14102

Are you sure you want to change the base?

PIP-144: Making SchemaRegistry implementation configurable #14102

Conversation

aparajita89 commented Feb 3, 2022

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aparajita89 commented Feb 9, 2022

Huanli-Meng commented Feb 10, 2022

aparajita89 commented Feb 10, 2022

aparajita89 commented Feb 10, 2022

Anonymitaet commented Feb 10, 2022

aparajita89 commented Feb 11, 2022

Huanli-Meng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Huanli-Meng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

aparajita89 commented Feb 17, 2022

aparajita89 commented Feb 21, 2022

aparajita89 commented Apr 19, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Apr 21, 2022

codelipenghui commented Apr 25, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jun 18, 2022