[PIP-44] Separate schema compatibility checker for producer and consumer

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -81,12 +81,14 @@
 import org.apache.pulsar.common.policies.data.PublishRate;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.SubscribeRate;
 import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.data.Stat;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -2034,11 +2036,18 @@ protected void internalSetOffloadDeletionLag(Long newDeletionLagMs) {
         }
     }
 
+    @Deprecated
     protected SchemaAutoUpdateCompatibilityStrategy internalGetSchemaAutoUpdateCompatibilityStrategy() {
         validateAdminAccessForTenant(namespaceName.getTenant());
         return getNamespacePolicies(namespaceName).schema_auto_update_compatibility_strategy;
     }
 
+    protected SchemaCompatibilityStrategy internalGetSchemaCompatibilityStrategy() {
+        validateAdminAccessForTenant(namespaceName.getTenant());
+        return getNamespacePolicies(namespaceName).schema_compatibility_strategy;
+    }
+
+    @Deprecated
     protected void internalSetSchemaAutoUpdateCompatibilityStrategy(SchemaAutoUpdateCompatibilityStrategy strategy) {
         validateSuperUserAccess();
         validatePoliciesReadOnlyAccess();
@@ -2050,6 +2059,17 @@ protected void internalSetSchemaAutoUpdateCompatibilityStrategy(SchemaAutoUpdate
             "schemaAutoUpdateCompatibilityStrategy");
     }
 
+    protected void internalSetSchemaCompatibilityStrategy(SchemaCompatibilityStrategy strategy) {
+        validateSuperUserAccess();
+        validatePoliciesReadOnlyAccess();
+
+        mutatePolicy((policies) -> {
+                    policies.schema_compatibility_strategy = strategy;
+                    return policies;
+                }, (policies) -> policies.schema_compatibility_strategy,
+                "schemaCompatibilityStrategy");
+    }
+
     protected boolean internalGetSchemaValidationEnforced() {
         validateSuperUserAccess();
         validateAdminAccessForTenant(namespaceName.getTenant());
@@ -2067,6 +2087,23 @@ protected void internalSetSchemaValidationEnforced(boolean schemaValidationEnfor
             "schemaValidationEnforced");
     }
 
+    protected boolean internalGetIsAllowAutoUpdateSchema() {
+        validateSuperUserAccess();
+        validateAdminAccessForTenant(namespaceName.getTenant());
+        return getNamespacePolicies(namespaceName).is_allow_auto_update_schema;
+    }
+
+    protected void internalSetIsAllowAutoUpdateSchema(boolean isAllowAutoUpdateSchema) {
+        validateSuperUserAccess();
+        validatePoliciesReadOnlyAccess();
+
+        mutatePolicy((policies) -> {
+                    policies.is_allow_auto_update_schema = isAllowAutoUpdateSchema;
+                    return policies;
+                }, (policies) -> policies.is_allow_auto_update_schema,
+                "isAllowAutoUpdateSchema");
+    }
+
 
     private <T> void mutatePolicy(Function<Policies, Policies> policyTransformation,
                                   Function<Policies, T> getter,

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -55,9 +55,11 @@
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.PublishRate;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.SubscribeRate;
 import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -965,6 +967,59 @@ public void setSchemaAutoUpdateCompatibilityStrategy(@PathParam("tenant") String
         internalSetSchemaAutoUpdateCompatibilityStrategy(strategy);
     }
 
+    @GET
+    @Path("/{tenant}/{namespace}/schemaCompatibilityStrategy")
+    @ApiOperation(value = "The strategy of the namespace schema compatibility ")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Namespace doesn't exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification") })
+    public SchemaCompatibilityStrategy getSchemaCompatibilityStrategy(
+            @PathParam("tenant") String tenant,
+            @PathParam("namespace") String namespace) {
+        validateNamespaceName(tenant, namespace);
+        return internalGetSchemaCompatibilityStrategy();
+    }
+
+    @PUT
+    @Path("/{tenant}/{namespace}/schemaCompatibilityStrategy")
+    @ApiOperation(value = "Update the strategy used to check the compatibility of new schema")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Namespace doesn't exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification") })
+    public void setSchemaCompatibilityStrategy(@PathParam("tenant") String tenant,
+                                               @PathParam("namespace") String namespace,
+                                               SchemaCompatibilityStrategy strategy) {
+        validateNamespaceName(tenant, namespace);
+        internalSetSchemaCompatibilityStrategy(strategy);
+    }
+
+    @GET
+    @Path("/{tenant}/{namespace}/isAllowAutoUpdateSchema")
+    @ApiOperation(value = "The flag of whether allow auto update schema")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Namespace doesn't exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification") })
+    public boolean getIsAllowAutoUpdateSchema(
+            @PathParam("tenant") String tenant,
+            @PathParam("namespace") String namespace) {
+        validateNamespaceName(tenant, namespace);
+        return internalGetIsAllowAutoUpdateSchema();
+    }
+
+    @POST
+    @Path("/{tenant}/{namespace}/isAllowAutoUpdateSchema")
+    @ApiOperation(value = "Update flag of whether allow auto update schema")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Namespace doesn't exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification") })
+    public void setIsAllowAutoUpdateSchema(@PathParam("tenant") String tenant,
+                                           @PathParam("namespace") String namespace,
+                                           boolean isAllowAutoUpdateSchema) {
+        validateNamespaceName(tenant, namespace);
+        internalSetIsAllowAutoUpdateSchema(isAllowAutoUpdateSchema);
+    }
+
+
     @GET
     @Path("/{tenant}/{namespace}/schemaValidationEnforced")
     @ApiOperation(value = "Get schema validation enforced flag for namespace.",

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java

@@ -55,7 +55,8 @@
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
 import org.apache.pulsar.broker.service.schema.LongSchemaVersion;
-import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.policies.data.Policies;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata;
 import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
 import org.apache.pulsar.broker.web.RestException;
@@ -314,34 +315,37 @@ public void postSchema(
 
         NamespaceName namespaceName = NamespaceName.get(tenant, namespace);
         getNamespacePoliciesAsync(namespaceName).thenAccept(policies -> {
-            SchemaCompatibilityStrategy schemaCompatibilityStrategy = SchemaCompatibilityStrategy
-                    .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
-        byte[] data;
-        if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
-            data = DefaultImplementation
-                    .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
-        } else {
-            data = payload.getSchema().getBytes(Charsets.UTF_8);
-        }
-        pulsar().getSchemaRegistryService().putSchemaIfAbsent(
-            buildSchemaId(tenant, namespace, topic),
-            SchemaData.builder()
-                .data(data)
-                .isDeleted(false)
-                .timestamp(clock.millis())
-                .type(SchemaType.valueOf(payload.getType()))
-                .user(defaultIfEmpty(clientAppId(), ""))
-                .props(payload.getProperties())
-                .build(),
+            SchemaCompatibilityStrategy schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
+            if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
+                schemaCompatibilityStrategy = SchemaCompatibilityStrategy
+                        .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
+            }
+            byte[] data;
+            if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
+                data = DefaultImplementation
+                        .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
+            } else {
+                data = payload.getSchema().getBytes(Charsets.UTF_8);
+            }
+            pulsar().getSchemaRegistryService().putSchemaIfAbsent(
+                buildSchemaId(tenant, namespace, topic),
+                SchemaData.builder()
+                    .data(data)
+                    .isDeleted(false)
+                    .timestamp(clock.millis())
+                    .type(SchemaType.valueOf(payload.getType()))
+                    .user(defaultIfEmpty(clientAppId(), ""))
+                    .props(payload.getProperties())
+                    .build(),
                 schemaCompatibilityStrategy
-        ).thenAccept(version ->
-            response.resume(
-                Response.accepted().entity(
-                    PostSchemaResponse.builder()
-                        .version(version)
-                        .build()
-                ).build()
-            )
+            ).thenAccept(version ->
+                    response.resume(
+                            Response.accepted().entity(
+                                    PostSchemaResponse.builder()
+                                            .version(version)
+                                            .build()
+                            ).build()
+                    )
         ).exceptionally(error -> {
             if (error.getCause() instanceof IncompatibleSchemaException) {
                 response.resume(Response.status(Response.Status.CONFLICT.getStatusCode(),
@@ -407,10 +411,15 @@ public void testCompatibility(
         validateDestinationAndAdminOperation(tenant, namespace, topic, authoritative);
 
         String schemaId = buildSchemaId(tenant, namespace, topic);
+        Policies policies = getNamespacePolicies(NamespaceName.get(tenant, namespace));
 
-        SchemaCompatibilityStrategy schemaCompatibilityStrategy = SchemaCompatibilityStrategy
-                .fromAutoUpdatePolicy(getNamespacePolicies(NamespaceName.get(tenant, namespace))
-                        .schema_auto_update_compatibility_strategy);
+        SchemaCompatibilityStrategy schemaCompatibilityStrategy;
+        if (policies.schema_compatibility_strategy == SchemaCompatibilityStrategy.UNDEFINED) {
+            schemaCompatibilityStrategy = SchemaCompatibilityStrategy
+                    .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
+        } else {
+            schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
+        }
 
         pulsar().getSchemaRegistryService().isCompatible(schemaId, SchemaData.builder()
                         .data(payload.getSchema().getBytes(Charsets.UTF_8))

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -28,14 +28,16 @@
 
 import org.apache.bookkeeper.mledger.util.StatsBuckets;
 import org.apache.pulsar.broker.admin.AdminResource;
-import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy;
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
 import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.PublishRate;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
@@ -75,6 +77,7 @@ public abstract class AbstractTopic implements Topic {
     protected volatile boolean isEncryptionRequired = false;
     protected volatile SchemaCompatibilityStrategy schemaCompatibilityStrategy =
             SchemaCompatibilityStrategy.FULL;
+    protected volatile boolean isAllowAutoUpdateSchema = true;
     // schema validation enforced flag
     protected volatile boolean schemaValidationEnforced = false;
 
@@ -184,9 +187,13 @@ public CompletableFuture<SchemaVersion> addSchema(SchemaData schema) {
 
         String base = TopicName.get(getName()).getPartitionedTopicName();
         String id = TopicName.get(base).getSchemaName();
-        return brokerService.pulsar()
-                .getSchemaRegistryService()
-                .putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy);
+        if (isAllowAutoUpdateSchema) {
+            return brokerService.pulsar()
+                    .getSchemaRegistryService()
+                    .putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy);
+        } else {
+            return FutureUtil.failedFuture(new IncompatibleSchemaException("Don't allow auto update schema."));
+        }
     }
 
     @Override
@@ -205,12 +212,12 @@ public CompletableFuture<SchemaVersion> deleteSchema() {
     }
 
     @Override
-    public CompletableFuture<Boolean> isSchemaCompatible(SchemaData schema) {
+    public CompletableFuture<Void> checkSchemaCompatibleForConsumer(SchemaData schema) {
         String base = TopicName.get(getName()).getPartitionedTopicName();
         String id = TopicName.get(base).getSchemaName();
         return brokerService.pulsar()
                 .getSchemaRegistryService()
-                .isCompatible(id, schema, schemaCompatibilityStrategy);
+                .checkConsumerCompatibility(id, schema, schemaCompatibilityStrategy);
     }
 
     @Override
@@ -220,6 +227,14 @@ public void recordAddLatency(long latency, TimeUnit unit) {
         PUBLISH_LATENCY.observe(latency, unit);
     }
 
+    protected void setSchemaCompatibilityStrategy (Policies policies) {
+        if (policies.schema_compatibility_strategy == SchemaCompatibilityStrategy.UNDEFINED) {
+            schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
+                    policies.schema_auto_update_compatibility_strategy);
+        } else {
+            schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
+        }
+    }
     private static final Summary PUBLISH_LATENCY = Summary.build("pulsar_broker_publish_latency", "-")
             .quantile(0.0)
             .quantile(0.50)

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -34,7 +34,7 @@
 import io.netty.handler.ssl.SslHandler;
 
 import java.net.SocketAddress;
-import java.util.List;
+
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -703,19 +703,11 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
 
                                     if (schema != null) {
                                         return topic.addSchemaIfIdleOrCheckCompatible(schema)
-                                            .thenCompose(isCompatible -> {
-                                                    if (isCompatible) {
-                                                        return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
-                                                                subType, priorityLevel, consumerName, isDurable,
-                                                                startMessageId, metadata,
-                                                                readCompacted, initialPosition, startMessageRollbackDurationSec, isReplicated);
-                                                    } else {
-                                                        return FutureUtil.failedFuture(
-                                                                new IncompatibleSchemaException(
-                                                                        "Trying to subscribe with incompatible schema"
-                                                        ));
-                                                    }
-                                                });
+                                            .thenCompose(v -> topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
+                                                    subType, priorityLevel, consumerName, isDurable,
+                                                    startMessageId, metadata,
+                                                    readCompacted, initialPosition, startMessageRollbackDurationSec, isReplicated));
+
                                     } else {
                                         return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
                                             subType, priorityLevel, consumerName, isDurable,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -172,14 +172,14 @@ void updateRates(NamespaceStats nsStats, NamespaceBundleStats currentBundleStats
     /**
      * Check if schema is compatible with current topic schema.
      */
-    CompletableFuture<Boolean> isSchemaCompatible(SchemaData schema);
+    CompletableFuture<Void> checkSchemaCompatibleForConsumer(SchemaData schema);
 
     /**
      * If the topic is idle (no producers, no entries, no subscribers and no existing schema),
      * add the passed schema to the topic. Otherwise, check that the passed schema is compatible
      * with what the topic already has.
      */
-    CompletableFuture<Boolean> addSchemaIfIdleOrCheckCompatible(SchemaData schema);
+    CompletableFuture<Void> addSchemaIfIdleOrCheckCompatible(SchemaData schema);
 
     CompletableFuture<Void> deleteForcefully();