congbobo184 commented on a change in pull request #13297:
URL: https://github.com/apache/pulsar/pull/13297#discussion_r781934103
##########
File path:
pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java
##########
@@ -170,4 +171,8 @@ public boolean isSubscribeRateSet() {
public Set<String> getReplicationClustersSet() {
return replicationClusters != null ?
Sets.newTreeSet(this.replicationClusters) : null;
}
+
+ public boolean isSchemaCompatibilityStrategySet() {
Review comment:
seem not use this method
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaTypeCompatibilityCheckTest.java
##########
@@ -35,33 +42,24 @@
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.schema.Schemas;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.util.Collections;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.expectThrows;
-
-
public class SchemaTypeCompatibilityCheckTest extends
MockedPulsarServiceBaseTest {
private static final String CLUSTER_NAME = "test";
private static final String PUBLIC_TENANT = "public";
private static final String namespace = "test-namespace";
private static final String namespaceName = PUBLIC_TENANT + "/" +
namespace;
- @BeforeClass
+ @BeforeMethod
Review comment:
why change this
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -181,11 +211,48 @@ protected void
updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) {
topicPolicies.getDeduplicationEnabled().updateNamespaceValue(namespacePolicies.deduplicationEnabled);
topicPolicies.getSubscriptionTypesEnabled().updateNamespaceValue(
subTypeStringsToEnumSet(namespacePolicies.subscription_types_enabled));
+
+ updateSchemaCompatibilityStrategyNamespaceValue(namespacePolicies);
+
Arrays.stream(BacklogQuota.BacklogQuotaType.values()).forEach(
type -> this.topicPolicies.getBackLogQuotaMap().get(type)
.updateNamespaceValue(MapUtils.getObject(namespacePolicies.backlog_quota_map,
type)));
}
+ private void updateSchemaCompatibilityStrategyNamespaceValue(Policies
namespacePolicies){
+ SchemaCompatibilityStrategy schemaCompatibilityStrategy =
+
formatSchemaCompatibilityStrategy(namespacePolicies.schema_compatibility_strategy);
+ if (this.isSystemTopic()) {
+
topicPolicies.getSchemaCompatibilityStrategy().updateNamespaceValue(schemaCompatibilityStrategy);
+ return;
+ }
+
topicPolicies.getSchemaCompatibilityStrategy().updateNamespaceValue(schemaCompatibilityStrategy);
+
+ // If the broker config is null, use
schema_auto_update_compatibility_strategy from namespace policies as the
+ // broker value.
+ // This is the initial broker config.
+ SchemaCompatibilityStrategy brokerSchemaCompatibilityStrategyFromCache
=
+
topicPolicies.getSchemaCompatibilityStrategy().getBrokerValue();
+ SchemaCompatibilityStrategy
brokerSchemaCompatibilityStrategyFromPulsarConfig =
+
brokerService.pulsar().getConfiguration().getSchemaCompatibilityStrategy();
+ if (brokerSchemaCompatibilityStrategyFromPulsarConfig ==
SchemaCompatibilityStrategy.UNDEFINED) {
+ brokerSchemaCompatibilityStrategyFromPulsarConfig = null;
+ }
+
+ // When the initial broker config is null or the initial broker config
is not equal to the current broker
Review comment:
what time does the broker config is not equal to the current broker
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
##########
@@ -3491,5 +3492,114 @@ 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,
+ @ApiParam(value = "Is authentication required to perform this
operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean
authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+
+ preValidation(authoritative)
+ .thenCompose(__-> getSchemaCompatibilityStrategyAsync())
Review comment:
if don't add applied, we may should return the topicPolicy
`SchemaCompatibilityStrateg`, although it may be null
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -164,6 +171,29 @@ protected void updateTopicPolicy(TopicPolicies data) {
data.getBackLogQuotaMap() == null ? null :
data.getBackLogQuotaMap().get(type.toString())));
topicPolicies.getTopicMaxMessageSize().updateTopicValue(data.getMaxMessageSize());
topicPolicies.getMessageTTLInSeconds().updateTopicValue(data.getMessageTTLInSeconds());
+
updateSchemaCompatibilityStrategyTopicValue(data.getSchemaCompatibilityStrategy());
Review comment:
topicPolicies.getSchemaCompatibilityStrategy()
.updateTopicValue(formatSchemaCompatibilityStrategy(config.getSchemaCompatibilityStrategy()));
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
