diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index aeb8e1e2ab0b8..85d043f09856c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/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 void mutatePolicy(Function policyTransformation, Function getter, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java index e80ae2d39eca5..dfa0ce95c59f1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java +++ b/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.", diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java index b91e577a0d38e..868c7ec90ab60 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java +++ b/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)) 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 d3bbb7ded02b2..406704a1c6cf7 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 @@ -28,7 +28,8 @@ 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; @@ -36,6 +37,7 @@ 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 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 deleteSchema() { } @Override - public CompletableFuture isSchemaCompatible(SchemaData schema) { + public CompletableFuture 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) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 4c2e0eed83917..cb4e203d54732 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/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, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java index 5adaf2a319c98..53a33ea4b47ce 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java +++ b/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 isSchemaCompatible(SchemaData schema); + CompletableFuture 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 addSchemaIfIdleOrCheckCompatible(SchemaData schema); + CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema); CompletableFuture deleteForcefully(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 870525b7e77f0..9ea86b25a71d3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -60,7 +60,7 @@ import org.apache.pulsar.broker.service.StreamingStats; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.Topic; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.stats.NamespaceStats; import org.apache.pulsar.client.api.MessageId; @@ -144,8 +144,9 @@ public NonPersistentTopic(String topic, BrokerService brokerService) { .get(AdminResource.path(POLICIES, TopicName.get(topic).getNamespace())) .orElseThrow(() -> new KeeperException.NoNodeException()); isEncryptionRequired = policies.encryption_required; - schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy( - policies.schema_auto_update_compatibility_strategy); + isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema; + setSchemaCompatibilityStrategy(policies); + schemaValidationEnforced = policies.schema_validation_enforced; } catch (Exception e) { @@ -870,8 +871,8 @@ public CompletableFuture onPoliciesUpdate(Policies data) { log.debug("[{}] isEncryptionRequired changes: {} -> {}", topic, isEncryptionRequired, data.encryption_required); } isEncryptionRequired = data.encryption_required; - schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy( - data.schema_auto_update_compatibility_strategy); + setSchemaCompatibilityStrategy(data); + isAllowAutoUpdateSchema = data.is_allow_auto_update_schema; schemaValidationEnforced = data.schema_validation_enforced; producers.forEach(producer -> { @@ -922,13 +923,14 @@ public Position getLastMessageId() { @Override - public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { + public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { return hasSchema() - .thenCompose((hasSchema) -> { + .thenCompose((hasSchema) -> { if (hasSchema || isActive() || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 0) { - return isSchemaCompatible(schema); + return checkSchemaCompatibleForConsumer(schema); } else { - return addSchema(schema).thenApply((ignore) -> true); + return addSchema(schema).thenCompose(schemaVersion-> + CompletableFuture.completedFuture(null)); } }); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 3da39b256fe60..13b213ac251d4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -19,30 +19,26 @@ package org.apache.pulsar.broker.service.persistent; import com.carrotsearch.hppc.ObjectObjectHashMap; + import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.common.collect.Sets; + import io.netty.buffer.ByteBuf; import io.netty.util.concurrent.FastThreadLocal; -import static com.google.common.base.Preconditions.checkArgument; -import static com.google.common.base.Preconditions.checkNotNull; -import static org.apache.commons.lang3.StringUtils.isBlank; -import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLongFieldUpdater; -import java.util.function.BiFunction; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback; @@ -90,7 +86,6 @@ import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.stats.NamespaceStats; import org.apache.pulsar.broker.stats.ReplicationMetrics; @@ -128,6 +123,15 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiFunction; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; + +import static org.apache.commons.lang3.StringUtils.isBlank; +import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES; + public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCallback { // Managed ledger associated with the topic @@ -238,8 +242,8 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS .orElseThrow(() -> new KeeperException.NoNodeException()); this.isEncryptionRequired = policies.encryption_required; - schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy( - policies.schema_auto_update_compatibility_strategy); + setSchemaCompatibilityStrategy(policies); + isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema; schemaValidationEnforced = policies.schema_validation_enforced; } catch (Exception e) { @@ -1683,12 +1687,11 @@ public CompletableFuture onPoliciesUpdate(Policies data) { } isEncryptionRequired = data.encryption_required; - schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy( - data.schema_auto_update_compatibility_strategy); + setSchemaCompatibilityStrategy(data); + isAllowAutoUpdateSchema = data.is_allow_auto_update_schema; schemaValidationEnforced = data.schema_validation_enforced; - initializeDispatchRateLimiterIfNeeded(Optional.ofNullable(data)); this.updateMaxPublishRate(data); @@ -1944,13 +1947,14 @@ public synchronized OffloadProcessStatus offloadStatus() { private static final Logger log = LoggerFactory.getLogger(PersistentTopic.class); @Override - public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { + public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schema) { return hasSchema() .thenCompose((hasSchema) -> { if (hasSchema || isActive() || ledger.getTotalSize() != 0) { - return isSchemaCompatible(schema); + return checkSchemaCompatibleForConsumer(schema); } else { - return addSchema(schema).thenApply((ignore) -> true); + return addSchema(schema).thenCompose(schemaVersion -> + CompletableFuture.completedFuture(null)); } }); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java index 8f215cac21ccc..9b8d4a10d6a06 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service.schema; +import static com.google.common.base.Preconditions.checkArgument; import static java.nio.charset.StandardCharsets.UTF_8; import java.util.Collections; @@ -29,6 +30,7 @@ import org.apache.avro.SchemaValidator; import org.apache.avro.SchemaValidatorBuilder; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import lombok.extern.slf4j.Slf4j; @@ -46,6 +48,7 @@ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityS @Override public void checkCompatible(Iterable from, SchemaData to, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException { LinkedList fromList = new LinkedList<>(); + checkArgument(from != null, "check compatibility list is null"); try { for (SchemaData schemaData : from) { Schema.Parser parser = new Schema.Parser(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java index 78d3e8a238088..ae361db739ce9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java @@ -25,7 +25,6 @@ import static java.util.Objects.nonNull; import static java.util.concurrent.CompletableFuture.completedFuture; import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.Functions.newSchemaEntry; -import static org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl.NO_DELETED_VERSION; import com.google.common.annotations.VisibleForTesting; @@ -108,8 +107,8 @@ public void start() throws IOException { } @Override - public CompletableFuture put(String key, byte[] value, byte[] hash, long maxDeletedVersion) { - return putSchemaIfAbsent(key, value, hash, maxDeletedVersion).thenApply(LongSchemaVersion::new); + public CompletableFuture put(String key, byte[] value, byte[] hash) { + return putSchema(key, value, hash).thenApply(LongSchemaVersion::new); } @Override @@ -172,6 +171,7 @@ private CompletableFuture getSchema(String schemaId) { } SchemaStorageFormat.SchemaLocator schemaLocator = locator.get().locator; + return readSchemaEntry(schemaLocator.getInfo().getPosition()) .thenApply(entry -> new StoredSchema(entry.getSchemaData().toByteArray(), new LongSchemaVersion(schemaLocator.getInfo().getVersion()))); @@ -245,42 +245,6 @@ private CompletableFuture getSchema(String schemaId, long version) @NotNull private CompletableFuture putSchema(String schemaId, byte[] data, byte[] hash) { - return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry -> { - if (optLocatorEntry.isPresent()) { - // Schema locator was already present - return addNewSchemaEntryToStore(schemaId, optLocatorEntry.get().locator.getIndexList(), data) - .thenCompose(position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)); - } else { - // No schema was defined yet - CompletableFuture future = new CompletableFuture<>(); - createNewSchema(schemaId, data, hash) - .thenAccept(version -> future.complete(version)) - .exceptionally(ex -> { - if (ex.getCause() instanceof NodeExistsException) { - // There was a race condition on the schema creation. Since it has now been created, - // retry the whole operation so that we have a chance to recover without bubbling error - // back to producer/consumer - putSchema(schemaId, data, hash) - .thenAccept(version -> future.complete(version)) - .exceptionally(ex2 -> { - future.completeExceptionally(ex2); - return null; - }); - } else { - // For other errors, just fail the operation - future.completeExceptionally(ex); - } - - return null; - }); - - return future; - } - }); - } - - @NotNull - private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, byte[] hash, long maxDeletedVersion) { return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry -> { if (optLocatorEntry.isPresent()) { @@ -295,26 +259,22 @@ private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, log.debug("[{}] findSchemaEntryByHash - hash={}", schemaId, hash); } - return findSchemaEntryByHash(locator.getIndexList(), hash, maxDeletedVersion).thenCompose(version -> { - if (isNull(version)) { - return addNewSchemaEntryToStore(schemaId, locator.getIndexList(), data).thenCompose( - position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)); - } else { - return completedFuture(version); - } - }); + //don't check the schema whether already exist + return readSchemaEntry(locator.getIndexList().get(0).getPosition()) + .thenCompose(schemaEntry -> addNewSchemaEntryToStore(schemaId, locator.getIndexList(), data).thenCompose( + position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash))); } else { // No schema was defined yet CompletableFuture future = new CompletableFuture<>(); createNewSchema(schemaId, data, hash) - .thenAccept(version -> future.complete(version)) + .thenAccept(future::complete) .exceptionally(ex -> { if (ex.getCause() instanceof NodeExistsException) { // There was a race condition on the schema creation. Since it has now been created, // retry the whole operation so that we have a chance to recover without bubbling error // back to producer/consumer - putSchemaIfAbsent(schemaId, data, hash, NO_DELETED_VERSION) - .thenAccept(version -> future.complete(version)) + putSchema(schemaId, data, hash) + .thenAccept(future::complete) .exceptionally(ex2 -> { future.completeExceptionally(ex2); return null; @@ -323,7 +283,6 @@ private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, // For other errors, just fail the operation future.completeExceptionally(ex); } - return null; }); @@ -439,36 +398,6 @@ private CompletableFuture findSchemaEntryByVers return completedFuture(null); } - @NotNull - private CompletableFuture findSchemaEntryByHash( - List index, - byte[] hash, - long maxDeletedVersion - ) { - - if (index.isEmpty()) { - return completedFuture(null); - } - - for (SchemaStorageFormat.IndexEntry entry : index) { - if (Arrays.equals(entry.getHash().toByteArray(), hash)) { - if (entry.getVersion() > maxDeletedVersion) { - return completedFuture(entry.getVersion()); - } else { - return completedFuture(null); - } - } - } - - if (index.get(0).getPosition().getLedgerId() == -1) { - return completedFuture(null); - } else { - return readSchemaEntry(index.get(0).getPosition()) - .thenCompose(entry -> findSchemaEntryByHash(entry.getIndexList(), hash, maxDeletedVersion)); - } - - } - @NotNull private CompletableFuture readSchemaEntry( SchemaStorageFormat.PositionInfo position diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/DefaultSchemaRegistryService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/DefaultSchemaRegistryService.java index c4f4d94e3f6f2..8bbda1119990d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/DefaultSchemaRegistryService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/DefaultSchemaRegistryService.java @@ -23,6 +23,8 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; + +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.protocol.schema.SchemaVersion; @@ -42,6 +44,11 @@ public CompletableFuture>> getAllSchem return completedFuture(Collections.emptyList()); } + @Override + public CompletableFuture putSchemaIfAbsent(String schemaId, SchemaData schema, SchemaCompatibilityStrategy strategy) { + return completedFuture(null); + } + @Override public CompletableFuture> trimDeletedSchemaAndGetList(String schemaId) { return completedFuture(Collections.emptyList()); @@ -52,10 +59,8 @@ public CompletableFuture findSchemaVersion(String schemaId, SchemaData sch return completedFuture(NO_SCHEMA_VERSION); } - @Override - public CompletableFuture putSchemaIfAbsent(String schemaId, SchemaData schema, - SchemaCompatibilityStrategy strategy) { + public CompletableFuture checkConsumerCompatibility(String schemaId, SchemaData schemaData, SchemaCompatibilityStrategy strategy) { return completedFuture(null); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java index beb78d5301c85..8f4f1398446b2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java @@ -26,6 +26,7 @@ import org.apache.avro.Schema; import org.apache.avro.SchemaParseException; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java index 006b63cbed64f..8d0487c316076 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java @@ -20,6 +20,7 @@ import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; import org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaInfo; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityCheck.java index 800c2afc392ad..9b6ffccf04000 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityCheck.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityCheck.java @@ -19,11 +19,10 @@ package org.apache.pulsar.broker.service.schema; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; -import java.util.Collections; - public interface SchemaCompatibilityCheck { SchemaType getSchemaType(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistry.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistry.java index b5f93767770ba..5b041648952b1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistry.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistry.java @@ -23,6 +23,8 @@ import java.util.List; import java.util.Objects; import java.util.concurrent.CompletableFuture; + +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.protocol.schema.SchemaVersion; @@ -49,6 +51,9 @@ CompletableFuture checkCompatible(String schemaId, SchemaData schema, CompletableFuture findSchemaVersion(String schemaId, SchemaData schemaData); + CompletableFuture checkConsumerCompatibility(String schemaId, SchemaData schemaData, + SchemaCompatibilityStrategy strategy); + SchemaVersion versionFromBytes(byte[] version); class SchemaAndMetadata { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java index 25e2588147d5a..d4e395695d35e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java @@ -20,17 +20,21 @@ import static java.util.Objects.isNull; import static java.util.concurrent.CompletableFuture.completedFuture; -import static org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE; -import static org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy.FORWARD_TRANSITIVE; -import static org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy.FULL_TRANSITIVE; +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 static org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl.Functions.toMap; import static org.apache.pulsar.broker.service.schema.SchemaRegistryServiceImpl.Functions.toPairs; import com.google.common.annotations.VisibleForTesting; +import com.google.common.hash.HashFunction; +import com.google.common.hash.Hashing; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; + import java.time.Clock; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -42,16 +46,18 @@ import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; import org.apache.pulsar.broker.service.schema.proto.SchemaRegistryFormat; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.protocol.schema.SchemaHash; import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.common.protocol.schema.SchemaVersion; +import org.apache.pulsar.common.util.FutureUtil; public class SchemaRegistryServiceImpl implements SchemaRegistryService { + private static HashFunction hashFunction = Hashing.sha256(); private final Map compatibilityChecks; private final SchemaStorage schemaStorage; private final Clock clock; - protected static final long NO_DELETED_VERSION = -1L; @VisibleForTesting SchemaRegistryServiceImpl(SchemaStorage schemaStorage, Map compatibilityChecks, Clock clock) { @@ -106,39 +112,29 @@ public CompletableFuture>> getAllSchem @NotNull public CompletableFuture putSchemaIfAbsent(String schemaId, SchemaData schema, SchemaCompatibilityStrategy strategy) { - return getSchema(schemaId, SchemaVersion.Latest) - .thenCompose( - (existingSchema) -> - { - CompletableFuture maxDeleteVersionFuture; - if (existingSchema == null) { - maxDeleteVersionFuture = completedFuture(NO_DELETED_VERSION); - } else if (existingSchema.schema.isDeleted()) { - maxDeleteVersionFuture = completedFuture(((LongSchemaVersion)schemaStorage - .versionFromBytes(existingSchema.version.bytes())).getVersion()); - } else { - if (isTransitiveStrategy(strategy)) { - maxDeleteVersionFuture = checkCompatibilityWithAll(schemaId, schema, strategy); - - } else { - maxDeleteVersionFuture = new CompletableFuture<>(); - trimDeletedSchemaAndGetList(schemaId).thenAccept(schemaAndMetadataList -> { - checkCompatibilityWithLatest(schemaId, schema, strategy).whenComplete((v, ex) -> { - if (ex == null) { - Long maxDeleteVersion = ((LongSchemaVersion)schemaStorage - .versionFromBytes(schemaAndMetadataList.get(0).version.bytes())).getVersion() - 1L; - maxDeleteVersionFuture.complete(maxDeleteVersion); - } else { - maxDeleteVersionFuture.completeExceptionally(ex); - } - }); - }); - } - } - return maxDeleteVersionFuture; + return trimDeletedSchemaAndGetList(schemaId).thenCompose(schemaAndMetadataList -> { + final CompletableFuture completableFuture = new CompletableFuture<>(); + SchemaVersion schemaVersion; + for (SchemaAndMetadata schemaAndMetadata : schemaAndMetadataList) { + if (Arrays.equals(hashFunction.hashBytes(schemaAndMetadata.schema.getData()).asBytes(), + hashFunction.hashBytes(schema.getData()).asBytes())) { + schemaVersion = schemaAndMetadata.version; + completableFuture.complete(schemaVersion); + return completableFuture; } - ).thenCompose(maxDeleteVersion -> { - byte[] context = SchemaHash.of(schema).asBytes(); + } + CompletableFuture checkCompatibilityFurture = new CompletableFuture<>(); + if (schemaAndMetadataList.size() != 0) { + if (isTransitiveStrategy(strategy)) { + checkCompatibilityFurture = checkCompatibilityWithAll(schema, strategy, schemaAndMetadataList); + } else { + checkCompatibilityFurture = checkCompatibilityWithLatest(schemaId, schema, strategy); + } + } else { + checkCompatibilityFurture.complete(null); + } + return checkCompatibilityFurture.thenCompose(v -> { + byte[] context = hashFunction.hashBytes(schema.getData()).asBytes(); SchemaRegistryFormat.SchemaInfo info = SchemaRegistryFormat.SchemaInfo.newBuilder() .setType(Functions.convertFromDomainType(schema.getType())) .setSchema(ByteString.copyFrom(schema.getData())) @@ -148,18 +144,18 @@ public CompletableFuture putSchemaIfAbsent(String schemaId, Schem .setTimestamp(clock.millis()) .addAllProps(toPairs(schema.getProps())) .build(); - return schemaStorage.put(schemaId, info.toByteArray(), context, maxDeleteVersion); + return schemaStorage.put(schemaId, info.toByteArray(), context); + }); + + }); } @Override @NotNull public CompletableFuture deleteSchema(String schemaId, String user) { byte[] deletedEntry = deleted(schemaId, user).toByteArray(); - return trimDeletedSchemaAndGetList(schemaId).thenCompose(schemaAndMetadataList -> - schemaStorage.put(schemaId, deletedEntry, new byte[]{}, ((LongSchemaVersion)schemaStorage - .versionFromBytes(schemaAndMetadataList.get(0).version.bytes())).getVersion() - 1L)); - + return schemaStorage.put(schemaId, deletedEntry, new byte[]{}); } @Override @@ -183,7 +179,7 @@ public CompletableFuture checkCompatible(String schemaId, SchemaData schem case FORWARD_TRANSITIVE: case BACKWARD_TRANSITIVE: case FULL_TRANSITIVE: - return checkCompatibilityWithAll(schemaId, schema, strategy).thenApply(maxDeleteVersion -> null); + return checkCompatibilityWithAll(schemaId, schema, strategy); default: return checkCompatibilityWithLatest(schemaId, schema, strategy); } @@ -241,6 +237,25 @@ public CompletableFuture findSchemaVersion(String schemaId, SchemaData sch }); } + @Override + public CompletableFuture checkConsumerCompatibility(String schemaId, SchemaData schemaData, + SchemaCompatibilityStrategy strategy) { + return getSchema(schemaId).thenCompose(existingSchema -> { + if (existingSchema != null && !existingSchema.schema.isDeleted()) { + if (strategy == SchemaCompatibilityStrategy.BACKWARD || + strategy == SchemaCompatibilityStrategy.FORWARD || + strategy == SchemaCompatibilityStrategy.FORWARD_TRANSITIVE || + strategy == SchemaCompatibilityStrategy.FULL) { + return checkCompatibilityWithLatest(schemaId, schemaData, SchemaCompatibilityStrategy.BACKWARD); + } else { + return checkCompatibilityWithAll(schemaId, schemaData, strategy); + } + } else { + return FutureUtil.failedFuture(new IncompatibleSchemaException("Topic does not have schema to check")); + } + }); + } + private CompletableFuture checkCompatibilityWithLatest(String schemaId, SchemaData schema, SchemaCompatibilityStrategy strategy) { return getSchema(schemaId).thenCompose(existingSchema -> { @@ -259,22 +274,31 @@ private CompletableFuture checkCompatibilityWithLatest(String schemaId, Sc }); } - private CompletableFuture checkCompatibilityWithAll(String schemaId, SchemaData schema, + private CompletableFuture checkCompatibilityWithAll(String schemaId, SchemaData schema, SchemaCompatibilityStrategy strategy) { - return trimDeletedSchemaAndGetList(schemaId).thenCompose(schemaAndMetadataList -> { - CompletableFuture result = new CompletableFuture<>(); - try { - compatibilityChecks.getOrDefault(schema.getType(), SchemaCompatibilityCheck.DEFAULT).checkCompatible(schemaAndMetadataList - .stream() - .map(schemaAndMetadata -> schemaAndMetadata.schema) - .collect(Collectors.toList()), schema, strategy); - result.complete(((LongSchemaVersion)schemaStorage.versionFromBytes(schemaAndMetadataList.get(0).version.bytes())).getVersion()); - } catch (IncompatibleSchemaException e) { + return trimDeletedSchemaAndGetList(schemaId).thenCompose(schemaAndMetadataList -> + checkCompatibilityWithAll(schema, strategy, schemaAndMetadataList)); + } + + private CompletableFuture checkCompatibilityWithAll(SchemaData schema, + SchemaCompatibilityStrategy strategy, + List schemaAndMetadataList) { + CompletableFuture result = new CompletableFuture<>(); + try { + compatibilityChecks.getOrDefault(schema.getType(), SchemaCompatibilityCheck.DEFAULT).checkCompatible(schemaAndMetadataList + .stream() + .map(schemaAndMetadata -> schemaAndMetadata.schema) + .collect(Collectors.toList()), schema, strategy); + result.complete(null); + } catch (Exception e) { + if (e instanceof IncompatibleSchemaException) { result.completeExceptionally(e); + } else { + result.completeExceptionally(new IncompatibleSchemaException(e)); } - return result; - }); + } + return result; } public CompletableFuture> trimDeletedSchemaAndGetList(String schemaId) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorage.java index e59fa33231afa..f1336667bd5a1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaStorage.java @@ -24,7 +24,7 @@ public interface SchemaStorage { - CompletableFuture put(String key, byte[] value, byte[] hash, long maxDeletedVersion); + CompletableFuture put(String key, byte[] value, byte[] hash); CompletableFuture get(String key, SchemaVersion version); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidator.java index 9d23285680373..e589d06e09e48 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidator.java @@ -20,7 +20,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.service.schema.SchemaRegistryService; import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException; import org.apache.pulsar.common.protocol.schema.SchemaData; @@ -72,6 +72,11 @@ public CompletableFuture findSchemaVersion(String schemaId, SchemaData sch return this.service.findSchemaVersion(schemaId, schemaData); } + @Override + public CompletableFuture checkConsumerCompatibility(String schemaId, SchemaData schemaData, SchemaCompatibilityStrategy strategy) { + return this.service.checkConsumerCompatibility(schemaId, schemaData, strategy); + } + @Override public CompletableFuture putSchemaIfAbsent(String schemaId, SchemaData schema, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java index 5af5c7fced71a..ab1bd09db25fe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java @@ -28,7 +28,7 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.Topic; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java index bea4e91e6ed63..70881d69449d9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service.schema; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; import org.testng.Assert; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java index f234b18ee2235..4aa6dac4495de 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java @@ -35,6 +35,7 @@ import org.apache.pulsar.client.api.SchemaSerializationException; import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheckTest.java index 0755781e04323..d5912240b378e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheckTest.java @@ -27,6 +27,7 @@ import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.client.impl.schema.StringSchema; import org.apache.pulsar.client.impl.schema.KeyValueSchema; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; import org.testng.Assert; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java index 0c2a174f82825..e18fa839dec7b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java @@ -39,6 +39,7 @@ import com.google.common.hash.Hashing; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.common.protocol.schema.SchemaVersion; @@ -252,7 +253,6 @@ public void dontReAddExistingSchemaAtRoot() throws Exception { putSchema(schemaId1, schema1, version(0)); } - @Test public void trimDeletedSchemaAndGetListTest() throws Exception { List list = new ArrayList<>(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidatorTest.java index 885a9e980b61c..bc05957b085b0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaRegistryServiceWithSchemaDataValidatorTest.java @@ -30,7 +30,7 @@ import static org.testng.Assert.fail; import java.util.concurrent.CompletableFuture; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata; import org.apache.pulsar.broker.service.schema.SchemaRegistryService; import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java index 163185025a398..112bbe7cbe25a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java @@ -36,7 +36,7 @@ import lombok.Cleanup; -import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.service.schema.SchemaRegistry; import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException; import org.apache.pulsar.client.api.schema.GenericRecord; @@ -337,10 +337,9 @@ public void testAvroConsumerWithWrongPrestoredSchema() throws Exception { .build(), SchemaCompatibilityStrategy.FULL ).get(); - Consumer consumer = pulsarClient .newConsumer(AvroSchema.of(SchemaDefinition.builder(). - withPojo(AvroEncodedPojo.class).build())) + withPojo(AvroEncodedPojo.class).withAlwaysAllowNull(false).build())) .topic("persistent://my-property/use/my-ns/my-topic1") .subscriptionName("my-subscriber-name") .subscribe(); 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 new file mode 100644 index 0000000000000..f091d9a890f52 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java @@ -0,0 +1,370 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.schema.compatibility; + +import com.google.common.collect.Sets; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.util.Collections; +import java.util.concurrent.ThreadLocalRandom; + +import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT; +import static org.junit.Assert.assertEquals; + + +@Slf4j +public class SchemaCompatibilityCheckTest extends MockedPulsarServiceBaseTest { + private final static String CLUSTER_NAME = "test"; + + @BeforeMethod + @Override + public void setup() throws Exception { + super.internalSetup(); + + // Setup namespaces + admin.clusters().createCluster(CLUSTER_NAME, new ClusterData("http://127.0.0.1" + ":" + BROKER_WEBSERVICE_PORT)); + TenantInfo tenantInfo = new TenantInfo(); + tenantInfo.setAllowedClusters(Collections.singleton(CLUSTER_NAME)); + admin.tenants().createTenant(PUBLIC_TENANT, tenantInfo); + } + + @AfterMethod + @Override + public void cleanup() throws Exception { + super.internalCleanup(); + } + + @DataProvider(name = "CanReadLastSchemaCompatibilityStrategy") + public Object[] canReadLastSchemaCompatibilityStrategy() { + return new Object[] { + SchemaCompatibilityStrategy.BACKWARD, + SchemaCompatibilityStrategy.FORWARD_TRANSITIVE, + SchemaCompatibilityStrategy.FORWARD, + SchemaCompatibilityStrategy.FULL + }; + } + + @DataProvider(name = "ReadAllCheckSchemaCompatibilityStrategy") + public Object[] readAllCheckSchemaCompatibilityStrategy() { + return new Object[] { + SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE, + SchemaCompatibilityStrategy.FULL_TRANSITIVE + }; + } + + @DataProvider(name = "AllCheckSchemaCompatibilityStrategy") + public Object[] allCheckSchemaCompatibilityStrategy() { + return new Object[] { + SchemaCompatibilityStrategy.BACKWARD, + SchemaCompatibilityStrategy.FORWARD_TRANSITIVE, + SchemaCompatibilityStrategy.FORWARD, + SchemaCompatibilityStrategy.FULL, + SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE, + SchemaCompatibilityStrategy.FULL_TRANSITIVE + }; + } + + @Test(dataProvider = "CanReadLastSchemaCompatibilityStrategy") + public void testConsumerCompatibilityCheckCanReadLastTest(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) + ); + + admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy); + admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo()); + admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder() + .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo()); + + Consumer consumerThree = pulsarClient.newConsumer(Schema.AVRO( + SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonThree.class).build())) + .subscriptionName("test") + .topic(fqtn) + .subscribe(); + + Producer producerOne = pulsarClient + .newProducer(Schema.AVRO(Schemas.PersonOne.class)) + .topic(fqtn) + .create(); + + + Schemas.PersonOne personOne = new Schemas.PersonOne(); + personOne.id = 1; + + producerOne.send(personOne); + Message message = null; + + try { + message = consumerThree.receive(); + message.getValue(); + } catch (Exception e) { + Assert.assertTrue(e instanceof SchemaSerializationException); + consumerThree.acknowledge(message); + } + + Producer producerTwo = pulsarClient + .newProducer(Schema.AVRO(SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonTwo.class).build())) + .topic(fqtn) + .create(); + + Schemas.PersonTwo personTwo = new Schemas.PersonTwo(); + personTwo.id = 1; + personTwo.name = "Jerry"; + producerTwo.send(personTwo); + + message = consumerThree.receive(); + Schemas.PersonThree personThree = message.getValue(); + consumerThree.acknowledge(message); + + assertEquals(personThree.id, 1); + assertEquals(personThree.name, "Jerry"); + + consumerThree.close(); + producerOne.close(); + producerTwo.close(); + } + + @Test(dataProvider = "ReadAllCheckSchemaCompatibilityStrategy") + public void testConsumerCompatibilityReadAllCheckTest(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) + ); + + admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy); + admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo()); + admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder() + .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo()); + try { + pulsarClient.newConsumer(Schema.AVRO( + SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonThree.class).build())) + .subscriptionName("test") + .topic(fqtn) + .subscribe(); + + } catch (Exception e) { + Assert.assertTrue(e.getMessage().contains("Unable to read schema")); + } + } + + @Test(dataProvider = "AllCheckSchemaCompatibilityStrategy") + public void testIsAutoUpdateSchema(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) + ); + + admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy); + admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo()); + + admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), false); + ProducerBuilder producerThreeBuilder = pulsarClient + .newProducer(Schema.AVRO(SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonTwo.class).build())) + .topic(fqtn); + try { + producerThreeBuilder.create(); + } catch (Exception e) { + Assert.assertTrue(e.getMessage().contains("Don't allow auto update schema.")); + } + + admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), true); + + Producer producer = producerThreeBuilder.create(); + Consumer consumerTwo = pulsarClient.newConsumer(Schema.AVRO( + SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonTwo.class).build())) + .subscriptionName("test") + .topic(fqtn) + .subscribe(); + producer.send(new Schemas.PersonTwo(2, "Lucy")); + Message message = consumerTwo.receive(); + + Schemas.PersonTwo personTwo = message.getValue(); + consumerTwo.acknowledge(message); + + assertEquals(personTwo.id, 2); + assertEquals(personTwo.name, "Lucy"); + + consumerTwo.close(); + producer.close(); + } + + @Test(dataProvider = "AllCheckSchemaCompatibilityStrategy") + public void testProducerSendWithOldSchemaAndConsumerCanRead(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) + ); + + admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy); + admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo()); + admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder() + .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo()); + + Producer producerOne = pulsarClient + .newProducer(Schema.AVRO(Schemas.PersonOne.class)) + .topic(fqtn) + .create(); + + + Schemas.PersonOne personOne = new Schemas.PersonOne(10); + + Consumer consumerOne = pulsarClient.newConsumer(Schema.AVRO( + SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonOne.class).build())) + .subscriptionName("test") + .topic(fqtn) + .subscribe(); + + producerOne.send(personOne); + Message message = consumerOne.receive(); + personOne = message.getValue(); + + assertEquals(10, personOne.id); + + consumerOne.close(); + producerOne.close(); + + } + + @Test(dataProvider = "CanReadLastSchemaCompatibilityStrategy") + public void testConsumerWithNotCompatibilitySchema(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) + ); + + admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy); + admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo()); + admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder() + .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo()); + + try { + pulsarClient.newConsumer(Schema.AVRO( + SchemaDefinition.builder().withAlwaysAllowNull + (false).withSupportSchemaVersioning(true). + withPojo(Schemas.PersonFour.class).build())) + .subscriptionName("test") + .topic(fqtn) + .subscribe(); + } catch (Exception e) { + Assert.assertTrue(e.getMessage().contains("Unable to read schema")); + } + + } + public static String randomName(int numChars) { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < numChars; i++) { + sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a')); + } + return sb.toString(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/Schemas.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/Schemas.java new file mode 100644 index 0000000000000..6b63a25e5448a --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/Schemas.java @@ -0,0 +1,84 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.schema.compatibility; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.ToString; +import org.apache.avro.reflect.AvroDefault; + +public class Schemas { + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @NoArgsConstructor + @AllArgsConstructor + public static class PersonOne{ + int id; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor + @NoArgsConstructor + public static class PersonTwo{ + int id; + + @AvroDefault("\"Tom\"") + String name; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor + @NoArgsConstructor + public static class PersonThree{ + int id; + + String name; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor + @NoArgsConstructor + public static class PersonFour{ + int id; + + String name; + + int age; + } +} diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java index 9667ec3879f53..8b858ba53f93f 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Namespaces.java @@ -38,6 +38,7 @@ 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; @@ -1472,6 +1473,7 @@ CompletableFuture clearNamespaceBundleBacklogForSubscriptionAsync(String n * @throws PulsarAdminException * Unexpected error */ + @Deprecated SchemaAutoUpdateCompatibilityStrategy getSchemaAutoUpdateCompatibilityStrategy(String namespace) throws PulsarAdminException; @@ -1491,6 +1493,7 @@ SchemaAutoUpdateCompatibilityStrategy getSchemaAutoUpdateCompatibilityStrategy(S * @throws PulsarAdminException * Unexpected error */ + @Deprecated void setSchemaAutoUpdateCompatibilityStrategy(String namespace, SchemaAutoUpdateCompatibilityStrategy strategy) throws PulsarAdminException; @@ -1505,7 +1508,6 @@ void setSchemaAutoUpdateCompatibilityStrategy(String namespace, * @throws PulsarAdminException * Unexpected error */ - boolean getSchemaValidationEnforced(String namespace) throws PulsarAdminException; /** @@ -1523,7 +1525,70 @@ boolean getSchemaValidationEnforced(String namespace) * @throws PulsarAdminException * Unexpected error */ - void setSchemaValidationEnforced(String namespace, boolean schemaValidationEnforced) throws PulsarAdminException; + + /** + * Get the strategy used to check the a new schema provided by a producer is compatible with the current schema + * before it is installed. + * + * @param namespace The namespace in whose policy we are interested + * @return the strategy used to check compatibility + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Namespace does not exist + * @throws PulsarAdminException + * Unexpected error + */ + SchemaCompatibilityStrategy getSchemaCompatibilityStrategy(String namespace) + throws PulsarAdminException;; + + /** + * Set the strategy used to check the a new schema provided by a producer is compatible with the current schema + * before it is installed. + * + * @param namespace The namespace in whose policy should be set + * @param strategy The schema compatibility strategy + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Namespace does not exist + * @throws PulsarAdminException + * Unexpected error + */ + void setSchemaCompatibilityStrategy(String namespace, + SchemaCompatibilityStrategy strategy) + throws PulsarAdminException; + + /** + * Get whether allow auto update schema + * + * @param namespace pulsar namespace name + * @return the schema validation enforced flag + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Tenant or Namespace does not exist + * @throws PulsarAdminException + * Unexpected error + */ + boolean getIsAllowAutoUpdateSchema(String namespace) + throws PulsarAdminException; + + /** + * The flag is when producer bring a new schema and the schema pass compatibility check + * whether allow schema auto registered + * + * @param namespace pulsar namespace name + * @param isAllowAutoUpdateSchema flag to enable or disable auto update schema + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Tenant or Namespace does not exist + * @throws PulsarAdminException + * Unexpected error + */ + void setIsAllowAutoUpdateSchema(String namespace, boolean isAllowAutoUpdateSchema) + throws PulsarAdminException; } diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java index dd4b4284388f4..7ab7762c7612d 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java @@ -48,6 +48,7 @@ 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; @@ -959,6 +960,52 @@ public void setSchemaValidationEnforced(String namespace, boolean schemaValidati } } + @Override + public SchemaCompatibilityStrategy getSchemaCompatibilityStrategy(String namespace) throws PulsarAdminException { + try { + NamespaceName ns = NamespaceName.get(namespace); + WebTarget path = namespacePath(ns, "schemaCompatibilityStrategy"); + return request(path).get(SchemaCompatibilityStrategy.class); + } catch (Exception e) { + throw getApiException(e); + } + } + + @Override + public void setSchemaCompatibilityStrategy(String namespace, SchemaCompatibilityStrategy strategy) throws PulsarAdminException { + try { + NamespaceName ns = NamespaceName.get(namespace); + WebTarget path = namespacePath(ns, "schemaCompatibilityStrategy"); + request(path).put(Entity.entity(strategy, MediaType.APPLICATION_JSON), + ErrorData.class); + } catch (Exception e) { + throw getApiException(e); + } + + } + + @Override + public boolean getIsAllowAutoUpdateSchema(String namespace) throws PulsarAdminException { + try { + NamespaceName ns = NamespaceName.get(namespace); + WebTarget path = namespacePath(ns, "isAllowAutoUpdateSchema"); + return request(path).get(Boolean.class); + } catch (Exception e) { + throw getApiException(e); + } + } + + @Override + public void setIsAllowAutoUpdateSchema(String namespace, boolean isAllowAutoUpdateSchema) throws PulsarAdminException { + try { + NamespaceName ns = NamespaceName.get(namespace); + WebTarget path = namespacePath(ns, "isAllowAutoUpdateSchema"); + request(path).post(Entity.entity(isAllowAutoUpdateSchema, MediaType.APPLICATION_JSON), ErrorData.class); + } catch (Exception e) { + throw getApiException(e); + } + } + private WebTarget namespacePath(NamespaceName namespace, String... parts) { final WebTarget base = namespace.isV2() ? adminV2Namespaces : adminNamespaces; WebTarget namespacePath = base.path(namespace.toString()); diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java index 4c7daea0d102b..eb72052dfd381 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java @@ -45,6 +45,7 @@ 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.RelativeTimeUtil; @@ -1130,6 +1131,77 @@ void run() throws PulsarAdminException { } } + @Parameters(commandDescription = "Get the schema compatibility strategy for a namespace") + private class GetSchemaCompatibilityStrategy extends CliCommand { + @Parameter(description = "tenant/namespace", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String namespace = validateNamespace(params); + System.out.println(admin.namespaces().getSchemaCompatibilityStrategy(namespace) + .toString().toUpperCase()); + } + } + + @Parameters(commandDescription = "Set the schema compatibility strategy for a namespace") + private class SetSchemaCompatibilityStrategy extends CliCommand { + @Parameter(description = "tenant/namespace", required = true) + private java.util.List params; + + @Parameter(names = { "--compatibility", "-c" }, + description = "Compatibility level required for new schemas created via a Producer. " + + "Possible values (FULL, BACKWARD, FORWARD, " + + "UNDEFINED, BACKWARD_TRANSITIVE, " + + "FORWARD_TRANSITIVE, FULL_TRANSITIVE, " + + "ALWAYS_INCOMPATIBLE," + + "ALWAYS_COMPATIBLE).") + private String strategyParam = null; + + @Override + void run() throws PulsarAdminException { + String namespace = validateNamespace(params); + + String strategyStr = strategyParam != null ? strategyParam.toUpperCase() : ""; + admin.namespaces().setSchemaCompatibilityStrategy(namespace, SchemaCompatibilityStrategy.valueOf(strategyStr)); + } + } + + @Parameters(commandDescription = "Get the namespace whether allow auto update schema") + private class GetIsAllowAutoUpdateSchema extends CliCommand { + @Parameter(description = "tenant/namespace", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String namespace = validateNamespace(params); + + System.out.println(admin.namespaces().getIsAllowAutoUpdateSchema(namespace)); + } + } + + @Parameters(commandDescription = "Set the namespace whether allow auto update schema") + private class SetIsAllowAutoUpdateSchema extends CliCommand { + @Parameter(description = "tenant/namespace", required = true) + private java.util.List params; + + @Parameter(names = { "--enable", "-e" }, description = "Enable schema validation enforced") + private boolean enable = false; + + @Parameter(names = { "--disable", "-d" }, description = "Disable schema validation enforced") + private boolean disable = false; + + @Override + void run() throws PulsarAdminException { + String namespace = validateNamespace(params); + + if (enable == disable) { + throw new ParameterException("Need to specify either --enable or --disable"); + } + admin.namespaces().setIsAllowAutoUpdateSchema(namespace, enable); + } + } + @Parameters(commandDescription = "Get the schema validation enforced") private class GetSchemaValidationEnforced extends CliCommand { @Parameter(description = "tenant/namespace", required = true) @@ -1256,6 +1328,12 @@ public CmdNamespaces(PulsarAdmin admin) { jcommander.addCommand("get-schema-autoupdate-strategy", new GetSchemaAutoUpdateStrategy()); jcommander.addCommand("set-schema-autoupdate-strategy", new SetSchemaAutoUpdateStrategy()); + jcommander.addCommand("get-schema-compatibility-strategy", new GetSchemaCompatibilityStrategy()); + jcommander.addCommand("set-schema-compatibility-strategy", new SetSchemaCompatibilityStrategy()); + + jcommander.addCommand("get-is-allow-auto-update-schema", new GetIsAllowAutoUpdateSchema()); + jcommander.addCommand("set-is-allow-auto-update-schema", new SetIsAllowAutoUpdateSchema()); + jcommander.addCommand("get-schema-validation-enforce", new GetSchemaValidationEnforced()); jcommander.addCommand("set-schema-validation-enforce", new SetSchemaValidationEnforced()); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java index e5334dfd1b6a0..6a0746c8af327 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java @@ -24,7 +24,6 @@ import java.net.InetSocketAddress; import java.net.URI; -import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Base64; import java.util.List; @@ -47,6 +46,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static com.yahoo.sketches.Util.bytesToLong; + class HttpLookupService implements LookupService { private final HttpClient httpClient; @@ -151,7 +152,7 @@ public CompletableFuture> getSchema(TopicName topicName, by if (version != null) { path = String.format("admin/v2/schemas/%s/schema/%s", schemaName, - new String(version, StandardCharsets.UTF_8)); + bytesToLong(version)); } httpClient.get(path, GetSchemaResponse.class).thenAccept(response -> { future.complete(Optional.of(SchemaInfoUtil.newSchemaInfo(schemaName, response))); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java index b38df410fa571..ede229cb74e63 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java @@ -28,11 +28,13 @@ import com.google.common.cache.LoadingCache; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufInputStream; +import org.apache.avro.AvroTypeException; import org.apache.avro.Schema.Parser; import org.apache.avro.reflect.ReflectData; import org.apache.commons.codec.binary.Hex; import org.apache.commons.lang3.SerializationException; import org.apache.commons.lang3.StringUtils; +import org.apache.pulsar.client.api.SchemaSerializationException; import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.api.schema.SchemaInfoProvider; import org.apache.pulsar.client.api.schema.SchemaReader; @@ -94,7 +96,10 @@ public T decode(byte[] bytes) { public T decode(byte[] bytes, byte[] schemaVersion) { try { return readerCache.get(BytesSchemaVersion.of(schemaVersion)).read(bytes); - } catch (ExecutionException e) { + } catch (ExecutionException | AvroTypeException e) { + if (e instanceof AvroTypeException) { + throw new SchemaSerializationException(e); + } LOG.error("Can't get generic schema for topic {} schema version {}", schemaInfoProvider.getTopicName(), Hex.encodeHexString(schemaVersion), e); throw new RuntimeException("Can't get generic schema for topic " + schemaInfoProvider.getTopicName()); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/AvroReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/AvroReader.java index dc5bfe6686b08..ad33ddecece46 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/AvroReader.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/AvroReader.java @@ -68,7 +68,7 @@ public T read(InputStream inputStream) { decoders.set(decoder); } return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, decoder)); - } catch (IOException e) { + } catch (Exception e) { throw new SchemaSerializationException(e); } finally { try { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java index fea610deb0c36..8e55280fc8de9 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java @@ -81,9 +81,16 @@ public class Policies { public Long offload_deletion_lag_ms = null; @SuppressWarnings("checkstyle:MemberName") + @Deprecated public SchemaAutoUpdateCompatibilityStrategy schema_auto_update_compatibility_strategy = SchemaAutoUpdateCompatibilityStrategy.Full; + @SuppressWarnings("checkstyle:MemberName") + public SchemaCompatibilityStrategy schema_compatibility_strategy = SchemaCompatibilityStrategy.UNDEFINED; + + @SuppressWarnings("checkstyle:MemberName") + public boolean is_allow_auto_update_schema = true; + @SuppressWarnings("checkstyle:MemberName") public boolean schema_validation_enforced = false; @@ -101,7 +108,9 @@ public int hashCode() { compaction_threshold, offload_threshold, offload_deletion_lag_ms, schema_auto_update_compatibility_strategy, - schema_validation_enforced); + schema_validation_enforced, + schema_compatibility_strategy, + is_allow_auto_update_schema); } @Override @@ -132,7 +141,9 @@ public boolean equals(Object obj) { && offload_threshold == other.offload_threshold && offload_deletion_lag_ms == other.offload_deletion_lag_ms && schema_auto_update_compatibility_strategy == other.schema_auto_update_compatibility_strategy - && schema_validation_enforced == other.schema_validation_enforced; + && schema_validation_enforced == other.schema_validation_enforced + && schema_compatibility_strategy == other.schema_compatibility_strategy + && is_allow_auto_update_schema == other.is_allow_auto_update_schema; } return false; @@ -178,6 +189,8 @@ public String toString() { .add("offload_threshold", offload_threshold) .add("offload_deletion_lag_ms", offload_deletion_lag_ms) .add("schema_auto_update_compatibility_strategy", schema_auto_update_compatibility_strategy) - .add("schema_validation_enforced", schema_validation_enforced).toString(); + .add("schema_validation_enforced", schema_validation_enforced) + .add("schema_compatibility_Strategy", schema_compatibility_strategy) + .add("is_allow_auto_update_Schema", is_allow_auto_update_schema).toString(); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityStrategy.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java similarity index 90% rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityStrategy.java rename to pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java index b15568d6f20c3..9a4f74c437b14 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaCompatibilityStrategy.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java @@ -16,33 +16,40 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.broker.service.schema; - -import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy; +package org.apache.pulsar.common.policies.data; +/** + * Pulsar Schema compatibility strategy. + */ public enum SchemaCompatibilityStrategy { + + /** + * Undefined. + */ + UNDEFINED, + /** - * Always incompatible + * Always incompatible. */ ALWAYS_INCOMPATIBLE, /** - * Always compatible + * Always compatible. */ ALWAYS_COMPATIBLE, /** - * Messages written by an old schema can be read by a new schema + * Messages written by an old schema can be read by a new schema. */ BACKWARD, /** - * Messages written by a new schema can be read by an old schema + * Messages written by a new schema can be read by an old schema. */ FORWARD, /** - * Equivalent to both FORWARD and BACKWARD + * Equivalent to both FORWARD and BACKWARD. */ FULL, @@ -59,7 +66,7 @@ public enum SchemaCompatibilityStrategy { FORWARD_TRANSITIVE, /** - * Equivalent to both FORWARD_TRANSITIVE and BACKWARD_TRANSITIVE + * Equivalent to both FORWARD_TRANSITIVE and BACKWARD_TRANSITIVE. */ FULL_TRANSITIVE; diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/SchemaTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/SchemaTest.java index 942748a0de08b..89e9291af0965 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/SchemaTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/SchemaTest.java @@ -268,7 +268,6 @@ public void testPrimitiveSchemaTypeCompatibilityCheck() { List schemas = new ArrayList<>(); schemas.add(Schema.STRING); - schemas.add(Schema.BYTES); schemas.add(Schema.INT8); schemas.add(Schema.INT16); schemas.add(Schema.INT32); @@ -279,35 +278,19 @@ public void testPrimitiveSchemaTypeCompatibilityCheck() { schemas.add(Schema.DATE); schemas.add(Schema.TIME); schemas.add(Schema.TIMESTAMP); - schemas.add(null); - - schemas.stream().forEach(schemaProducer -> { - schemas.stream().forEach(schemaConsumer -> { + schemas.forEach(schemaProducer -> { + schemas.forEach(schemaConsumer -> { try { String topicName = schemaProducer.getSchemaInfo().getName() + schemaConsumer.getSchemaInfo().getName(); - if (schemaProducer == null) { - client.newProducer() - .topic(topicName) - .create().close(); - } else { client.newProducer(schemaProducer) .topic(topicName) .create().close(); - } - if (schemaConsumer == null) { - client.newConsumer() - .topic(topicName) - .subscriptionName("test") - .subscribe().close(); - } else { client.newConsumer(schemaConsumer) .topic(topicName) .subscriptionName("test") .subscribe().close(); - } - assertEquals(schemaProducer.getSchemaInfo().getType(), schemaConsumer.getSchemaInfo().getType()); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/Schemas.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/Schemas.java index babb3c7858bc9..8158e107c6b33 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/Schemas.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/Schemas.java @@ -39,7 +39,9 @@ import lombok.NoArgsConstructor; import lombok.Setter; import lombok.ToString; + import org.apache.avro.reflect.AvroDefault; + import org.joda.time.DateTime; import org.joda.time.LocalDate; import org.joda.time.LocalTime; @@ -130,4 +132,53 @@ public static class AvroLogicalType{ private Schemas() {} + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor + @NoArgsConstructor + public static class PersonOne{ + int id; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor + @NoArgsConstructor + public static class PersonTwo{ + int id; + + @AvroDefault("\"Tom\"") + String name; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + public static class PersonThree{ + int id; + + String name; + } + + @Data + @Getter + @Setter + @ToString + @EqualsAndHashCode + public static class PersonFour{ + int id; + + String name; + + int age; + } + } diff --git a/tests/integration/src/test/resources/pulsar-schema-suite.xml b/tests/integration/src/test/resources/pulsar-schema-suite.xml new file mode 100644 index 0000000000000..83792268aecc7 --- /dev/null +++ b/tests/integration/src/test/resources/pulsar-schema-suite.xml @@ -0,0 +1,28 @@ + + + + + + + + diff --git a/tests/integration/src/test/resources/pulsar-schema.xml b/tests/integration/src/test/resources/pulsar-schema.xml new file mode 100644 index 0000000000000..7b3b21ef0dea6 --- /dev/null +++ b/tests/integration/src/test/resources/pulsar-schema.xml @@ -0,0 +1,28 @@ + + + + + + + + + diff --git a/tests/integration/src/test/resources/pulsar.xml b/tests/integration/src/test/resources/pulsar.xml index 468d3288ea4ce..e9339c43f760f 100644 --- a/tests/integration/src/test/resources/pulsar.xml +++ b/tests/integration/src/test/resources/pulsar.xml @@ -24,6 +24,7 @@ +