pip/pip-144.md
Pulsar comes with an inbuilt schema registry and a pluggable schema storage. Schema storage's default implementation is bookkeeper-based but a new implementation can be provided by the user. Schema registry on the other hand interacts with the configured schema storage and offers ability to do schema CRUD as well as performs compatibility checks.
Since schema registry is not pluggable, it is not possible as of today to enhance/extend what pulsar offers out of the box. This is usually true in organizations where an existing schema registry is already present and used in other systems that producers and consumers interact with. Having a pluggable storage is not sufficient, some of the reasons are:
For above reasons, we are proposing support for a pluggable schema registry in pulsar.
The scope of this is to make schema registry service pluggable in order to support different forms of versioning and compatibility checks, depending on the use case.
All of the changes are on the broker-side and backwards compatible. Existing schema registry service implementation will continue to remain the default one. Users will be able to provide a different registry by specifying the class name in the broker config and loading the relevant jar on the broker's class path. The experience for pluggable schema registry is going to be the same as the one for pluggable schema storage.
User-facing addition of flag in broker.conf
# Override the schema registry used by pulsar with a custom implementation. If this config is not provided,
# the default schema registry (SchemaRegistryServiceImpl) will be used.
schemaRegistryClassName=
Example of how a custom schema registry can be implemented by the user where they only want to change a certain set of behaviors and reuse the rest.
public class MySchemaRegistry extends SchemaRegistryServiceImpl {
@Override
public void initialize(ServiceConfiguration configuration, SchemaStorage schemaStorage) throws PulsarServerException {
super.initialize(configuration, schemaStorage);
//read config and do some other op
}
@Override
public CompletableFuture<List<CompletableFuture<SchemaAndMetadata>>> getAllSchemas(String schemaId) {
CompletableFuture<List<CompletableFuture<SchemaAndMetadata>>> schemas = super.getAllSchemas(schemaId);
//apply some filtering logic and return
}
@Override
public CompletableFuture<Boolean> isCompatible(String schemaId, SchemaData schema, SchemaCompatibilityStrategy strategy) {
//custom operation
}
}
Add a configuration parameter to set the schema registry to be used. This is illustrated in the "Goals" section above.
Add an initialize(...) method to SchemaRegistryService interface, similar in lines to the AuthorizationProvider interface already existing in Pulsar. This will ensure initialization of the plugged in schema registry with the required dependencies.
void initialize(ServiceConfiguration configuration, SchemaStorage schemaStorage) throws PulsarServerException;
These are the changes covered in the PR:
schemaRegistryClassName in ServiceConfiguration with a default value of org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImplinitialize(...) method instead of constructor to set dependencies like schema storage and compatibility checkers. This is being done to ensure that the contract becomes explicit to the implementer of the interface rather than having an assumption in the codebase around what the constructor params should be. The changes are:
initialize(...) method in SchemaRegistryService without an implementationcreate(...) method in SchemaRegistryService to create the instance of schema registry based on the ServiceConfiguration and initialize(...) method, instead of using a particular constructor overload.SchemaRegistryService interface to reflect their behavior. The changes are:
deleteSchema to putEmptySchema in SchemaRegistryServicedeleteSchemaStorage to deleteSchemaFromStorage in SchemaRegistryServicecheckCompatible from SchemaRegistryService and make it a private method in SchemaRegistryServiceImpl because it is not used anywhere elsecompatibilityChecks, schemaStorage and clock non-final in the default implementation as they are now being set in initialize(...) methodgetCheckers(...) from SchemaRegistryService to SchemaRegistryServiceImpl since this behavior is tied to how default schema registry implementation works.SchemaRegistryService to PulsarService. The null check will now be used to decide the name of the schema registry which needs to be instantiated.This change has been done and a PR has been raised here: https://github.com/apache/pulsar/pull/14102
SchemaStorage implementation itself (which is already pluggable and user-facing as of today). But the filtering needed to be done on the properties in SchemaData/SchemaInfo, and doing the deserialization in the schema storage layer did not match its purpose.Aparajita Singh (https://github.com/aparajita89) Ankur Jain (https://github.com/anvinjain)