Skip to content

Commit

Permalink
[Broker] Add schema compatibility strategy on topic level
Browse files Browse the repository at this point in the history 
### Motivation

Currently, the compatibility strategy for a schema is set at namespace level, which means that all the topics of a namespace will need to adhere to the same compatibility check, this level particle size is relatively large, so add schema compatibility strategy on topic level.

### Modifications

- Add set, get and remove schema compatibility strategy API to admin API and client tool
- Add schema compatibility strategy action to TopicOperation

### Verifying this change

- [x] Make sure that the change passes the CI checks.

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

*If `yes` was chosen, please highlight the changes*

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

### Documentation

- [x] `doc-required`

Signed-off-by: Zixuan Liu <[email protected]>
  • Loading branch information
@nodece
nodece committed Jan 27, 2022
1 parent 61f99cb commit 4053e29
Show file tree
Hide file tree
Showing 16 changed files with 786 additions and 51 deletions.
44 changes: 33 additions & 11 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -766,17 +766,39 @@ protected void resumeAsyncResponseExceptionally(AsyncResponse asyncResponse, Thr
}

protected CompletableFuture<SchemaCompatibilityStrategy> getSchemaCompatibilityStrategyAsync() {
return getNamespacePoliciesAsync(namespaceName).thenApply(policies -> {
SchemaCompatibilityStrategy schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
policies.schema_auto_update_compatibility_strategy);
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = pulsar().getConfig().getSchemaCompatibilityStrategy();
}
}
return schemaCompatibilityStrategy;
});
return validateTopicOperationAsync(topicName, TopicOperation.GET_SCHEMA_COMPATIBILITY_STRATEGY)
.thenCompose((__) -> {
CompletableFuture<SchemaCompatibilityStrategy> future;
if (config().isTopicLevelPoliciesEnabled()) {
future = getTopicPoliciesAsyncWithRetry(topicName)
.thenApply(op -> op.map(TopicPolicies::getSchemaCompatibilityStrategy).orElse(null));
} else {
future = CompletableFuture.completedFuture(null);
}

return future.thenCompose((topicSchemaCompatibilityStrategy) -> {
if (!SchemaCompatibilityStrategy.isUndefined(topicSchemaCompatibilityStrategy)) {
return CompletableFuture.completedFuture(topicSchemaCompatibilityStrategy);
}
return getNamespacePoliciesAsync(namespaceName).thenApply(policies -> {
SchemaCompatibilityStrategy schemaCompatibilityStrategy =
policies.schema_compatibility_strategy;
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
policies.schema_auto_update_compatibility_strategy);
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = pulsar().getConfig().getSchemaCompatibilityStrategy();
}
}
return schemaCompatibilityStrategy;
});
});
}).whenComplete((__, ex) -> {
if (ex != null) {
log.error("[{}] Failed to get schema compatibility strategy of topic {} {}",
clientAppId(), topicName, ex);
}
});
}

@CanIgnoreReturnValue
Expand Down
27 changes: 27 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@
import org.apache.pulsar.common.policies.data.PolicyOperation;
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.SubscribeRate;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TopicOperation;
Expand Down Expand Up @@ -4805,4 +4806,30 @@ private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(Async
resumeAsyncResponseExceptionally(asyncResponse, e);
}
}

protected CompletableFuture<SchemaCompatibilityStrategy> internalGetSchemaCompatibilityStrategy(boolean applied) {
if (applied) {
return getSchemaCompatibilityStrategyAsync();
}
return validateTopicOperationAsync(topicName, TopicOperation.GET_SCHEMA_COMPATIBILITY_STRATEGY)
.thenCompose(n -> getTopicPoliciesAsyncWithRetry(topicName).thenApply(op -> {
if (!op.isPresent()) {
return null;
}
SchemaCompatibilityStrategy strategy = op.get().getSchemaCompatibilityStrategy();
return SchemaCompatibilityStrategy.isUndefined(strategy) ? null : strategy;
}));
}

protected CompletableFuture<Void> internalSetSchemaCompatibilityStrategy(SchemaCompatibilityStrategy strategy) {
validateTopicOperation(topicName, TopicOperation.SET_SCHEMA_COMPATIBILITY_STRATEGY);

return getTopicPoliciesAsyncWithRetry(topicName)
.thenCompose(op -> {
TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
topicPolicies.setSchemaCompatibilityStrategy(
strategy == SchemaCompatibilityStrategy.UNDEFINED ? null : strategy);
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
});
}
}
111 changes: 111 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
import org.apache.pulsar.common.policies.data.PolicyOperation;
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.SubscribeRate;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicStats;
Expand Down Expand Up @@ -3566,5 +3567,115 @@ public void getReplicatedSubscriptionStatus(
internalGetReplicatedSubscriptionStatus(asyncResponse, decode(encodedSubName), authoritative);
}

@GET
@Path("/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy")
@ApiOperation(value = "Get schema compatibility strategy on a topic")
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 405, message = "Operation not allowed on persistent topic"),
@ApiResponse(code = 404, message = "Topic does not exist")})
public void getSchemaCompatibilityStrategy(
@Suspended AsyncResponse asyncResponse,
@ApiParam(value = "Specify the tenant", required = true)
@PathParam("tenant") String tenant,
@ApiParam(value = "Specify the cluster", required = true)
@PathParam("namespace") String namespace,
@ApiParam(value = "Specify topic name", required = true)
@PathParam("topic") @Encoded String encodedTopic,
@QueryParam("applied") boolean applied,
@ApiParam(value = "Is authentication required to perform this operation")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);

preValidation(authoritative)
.thenCompose(__-> internalGetSchemaCompatibilityStrategy(applied))
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
handleTopicPolicyException("getSchemaCompatibilityStrategy", ex, asyncResponse);
return null;
});
}

@PUT
@Path("/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy")
@ApiOperation(value = "Get schema compatibility strategy on a topic")
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 405, message = "Operation not allowed on persistent topic"),
@ApiResponse(code = 404, message = "Topic does not exist")})
public void setSchemaCompatibilityStrategy(
@Suspended AsyncResponse asyncResponse,
@ApiParam(value = "Specify the tenant", required = true)
@PathParam("tenant") String tenant,
@ApiParam(value = "Specify the namespace", required = true)
@PathParam("namespace") String namespace,
@ApiParam(value = "Specify topic name", required = true)
@PathParam("topic") @Encoded String encodedTopic,
@ApiParam(value = "Is authentication required to perform this operation")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@ApiParam(value = "Strategy used to check the compatibility of new schema")
SchemaCompatibilityStrategy strategy) {
validateTopicName(tenant, namespace, encodedTopic);

preValidation(authoritative)
.thenCompose(__ -> internalSetSchemaCompatibilityStrategy(strategy))
.thenRun(() -> {
log.info(
"[{}] Successfully set topic schema compatibility strategy: tenant={}, namespace={}, "
+ "topic={}, shemaCompatibilityStrategy = {}",
clientAppId(),
tenant,
namespace,
topicName.getLocalName(),
strategy);
asyncResponse.resume(Response.noContent().build());
}).exceptionally(ex -> {
handleTopicPolicyException("setSchemaCompatibilityStrategy", ex, asyncResponse);
return null;
});
}

@DELETE
@Path("/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy")
@ApiOperation(value = "Remove schema compatibility strategy on a topic")
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 405, message = "Operation not allowed on persistent topic"),
@ApiResponse(code = 404, message = "Topic does not exist")})
public void removeSchemaCompatibilityStrategy(
@Suspended AsyncResponse asyncResponse,
@ApiParam(value = "Specify the tenant", required = true)
@PathParam("tenant") String tenant,
@ApiParam(value = "Specify the namespace", required = true)
@PathParam("namespace") String namespace,
@ApiParam(value = "Specify topic name", required = true)
@PathParam("topic") @Encoded String encodedTopic,
@ApiParam(value = "Is authentication required to perform this operation")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@ApiParam(value = "Strategy used to check the compatibility of new schema")
SchemaCompatibilityStrategy strategy) {
validateTopicName(tenant, namespace, encodedTopic);

preValidation(authoritative)
.thenCompose(__ -> internalSetSchemaCompatibilityStrategy(null))
.thenRun(() -> {
log.info(
"[{}] Successfully remove topic schema compatibility strategy: tenant={}, namespace={}, "
+ "topic={}",
clientAppId(),
tenant,
namespace,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
handleTopicPolicyException("removeSchemaCompatibilityStrategy", ex, asyncResponse);
return null;
});
}

private static final Logger log = LoggerFactory.getLogger(PersistentTopics.class);
}
56 changes: 33 additions & 23 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,6 @@
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
Expand Down Expand Up @@ -104,9 +103,6 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
// Whether messages published must be encrypted or not in this topic
protected volatile boolean isEncryptionRequired = false;

@Getter
protected volatile SchemaCompatibilityStrategy schemaCompatibilityStrategy =
SchemaCompatibilityStrategy.FULL;
protected volatile Boolean isAllowAutoUpdateSchema;
// schema validation enforced flag
protected volatile boolean schemaValidationEnforced = false;
Expand Down Expand Up @@ -157,10 +153,20 @@ public AbstractTopic(String topic, BrokerService brokerService) {
updatePublishDispatcher(Optional.empty());
}

public SchemaCompatibilityStrategy getSchemaCompatibilityStrategy() {
return this.topicPolicies.getSchemaCompatibilityStrategy().get();
}

private SchemaCompatibilityStrategy formatSchemaCompatibilityStrategy(SchemaCompatibilityStrategy strategy) {
return strategy == SchemaCompatibilityStrategy.UNDEFINED ? null : strategy;
}

protected void updateTopicPolicy(TopicPolicies data) {
if (!isSystemTopic()) {
// Only use namespace level setting for system topic.
topicPolicies.getReplicationClusters().updateTopicValue(data.getReplicationClusters());
topicPolicies.getSchemaCompatibilityStrategy()
.updateTopicValue(formatSchemaCompatibilityStrategy(data.getSchemaCompatibilityStrategy()));
}
topicPolicies.getRetentionPolicies().updateTopicValue(data.getRetentionPolicies());
topicPolicies.getMaxSubscriptionsPerTopic().updateTopicValue(data.getMaxSubscriptionsPerTopic());
Expand Down Expand Up @@ -216,6 +222,21 @@ protected void updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) {
Arrays.stream(BacklogQuota.BacklogQuotaType.values()).forEach(
type -> this.topicPolicies.getBackLogQuotaMap().get(type)
.updateNamespaceValue(MapUtils.getObject(namespacePolicies.backlog_quota_map, type)));
updateSchemaCompatibilityStrategyNamespaceValue(namespacePolicies);
}

private void updateSchemaCompatibilityStrategyNamespaceValue(Policies namespacePolicies){
if (isSystemTopic()) {
return;
}

SchemaCompatibilityStrategy strategy = namespacePolicies.schema_compatibility_strategy;
if (SchemaCompatibilityStrategy.isUndefined(namespacePolicies.schema_compatibility_strategy)) {
strategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
namespacePolicies.schema_auto_update_compatibility_strategy);
}
topicPolicies.getSchemaCompatibilityStrategy()
.updateNamespaceValue(formatSchemaCompatibilityStrategy(strategy));
}

private void updateTopicPolicyByBrokerConfig() {
Expand Down Expand Up @@ -249,6 +270,12 @@ private void updateTopicPolicyByBrokerConfig() {
topicPolicies.getDelayedDeliveryTickTimeMillis().updateBrokerValue(config.getDelayedDeliveryTickTimeMillis());
topicPolicies.getCompactionThreshold().updateBrokerValue(config.getBrokerServiceCompactionThresholdInBytes());
topicPolicies.getReplicationClusters().updateBrokerValue(Collections.emptyList());
SchemaCompatibilityStrategy schemaCompatibilityStrategy = config.getSchemaCompatibilityStrategy();
if (isSystemTopic()) {
schemaCompatibilityStrategy = config.getSystemTopicSchemaCompatibilityStrategy();
}
topicPolicies.getSchemaCompatibilityStrategy()
.updateBrokerValue(formatSchemaCompatibilityStrategy(schemaCompatibilityStrategy));
}

private EnumSet<SubType> subTypeStringsToEnumSet(Set<String> getSubscriptionTypesEnabled) {
Expand Down Expand Up @@ -453,7 +480,7 @@ public CompletableFuture<SchemaVersion> addSchema(SchemaData schema) {
SchemaRegistryService schemaRegistryService = brokerService.pulsar().getSchemaRegistryService();

if (allowAutoUpdateSchema()) {
return schemaRegistryService.putSchemaIfAbsent(id, schema, schemaCompatibilityStrategy);
return schemaRegistryService.putSchemaIfAbsent(id, schema, getSchemaCompatibilityStrategy());
} else {
return schemaRegistryService.trimDeletedSchemaAndGetList(id).thenCompose(schemaAndMetadataList ->
schemaRegistryService.getSchemaVersionBySchemaData(schemaAndMetadataList, schema)
Expand Down Expand Up @@ -501,7 +528,7 @@ public CompletableFuture<Void> checkSchemaCompatibleForConsumer(SchemaData schem
String id = TopicName.get(base).getSchemaName();
return brokerService.pulsar()
.getSchemaRegistryService()
.checkConsumerCompatibility(id, schema, schemaCompatibilityStrategy);
.checkConsumerCompatibility(id, schema, getSchemaCompatibilityStrategy());
}

@Override
Expand Down Expand Up @@ -656,23 +683,6 @@ public long increasePublishLimitedTimes() {
return RATE_LIMITED_UPDATER.incrementAndGet(this);
}

protected void setSchemaCompatibilityStrategy(Policies policies) {
if (SystemTopicClient.isSystemTopic(TopicName.get(this.topic))) {
schemaCompatibilityStrategy =
brokerService.pulsar().getConfig().getSystemTopicSchemaCompatibilityStrategy();
return;
}

schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
policies.schema_auto_update_compatibility_strategy);
if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
schemaCompatibilityStrategy = brokerService.pulsar().getConfig().getSchemaCompatibilityStrategy();
}
}
}

private static final Summary PUBLISH_LATENCY = Summary.build("pulsar_broker_publish_latency", "-")
.quantile(0.0)
.quantile(0.50)
Expand Down
2 changes: 0 additions & 2 deletions ...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,6 @@ public CompletableFuture<Void> initialize() {
updateTopicPolicyByNamespacePolicy(policies);
isEncryptionRequired = policies.encryption_required;
isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema;
setSchemaCompatibilityStrategy(policies);
schemaValidationEnforced = policies.schema_validation_enforced;
}
});
Expand Down Expand Up @@ -1001,7 +1000,6 @@ public CompletableFuture<Void> onPoliciesUpdate(Policies data) {
updateTopicPolicyByNamespacePolicy(data);

isEncryptionRequired = data.encryption_required;
setSchemaCompatibilityStrategy(data);
isAllowAutoUpdateSchema = data.is_allow_auto_update_schema;
schemaValidationEnforced = data.schema_validation_enforced;

Expand Down
Loading

0 comments on commit 4053e29

Please sign in to comment.