From a4ba12d7764df6b6df9b902a971241a966d61baa Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 15 Feb 2022 23:26:42 +0800 Subject: [PATCH] [Broker] Move schema compatibility strategy cmd from topics to topicPolicies Signed-off-by: Zixuan Liu --- .../pulsar/admin/cli/CmdTopicPolicies.java | 51 ++++++++++++ .../apache/pulsar/admin/cli/CmdTopics.java | 49 ------------ .../SchemaCompatibilityStrategyTest.java | 78 ++++++++++--------- .../suites/PulsarCliTestSuite.java | 40 ++++++++++ .../topologies/PulsarClusterTestBase.java | 7 +- .../src/test/resources/pulsar-cli.xml | 1 + 6 files changed, 139 insertions(+), 87 deletions(-) create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarCliTestSuite.java diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopicPolicies.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopicPolicies.java index 00ea7fa2565a5..ba7d0ff77f0f8 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopicPolicies.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopicPolicies.java @@ -42,6 +42,7 @@ import org.apache.pulsar.common.policies.data.PersistencePolicies; 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.util.RelativeTimeUtil; @@ -141,6 +142,9 @@ public CmdTopicPolicies(Supplier admin) { jcommander.addCommand("set-max-subscriptions-per-topic", new SetMaxSubscriptionsPerTopic()); jcommander.addCommand("remove-max-subscriptions-per-topic", new RemoveMaxSubscriptionsPerTopic()); + jcommander.addCommand("remove-schema-compatibility-strategy", new RemoveSchemaCompatibilityStrategy()); + jcommander.addCommand("set-schema-compatibility-strategy", new SetSchemaCompatibilityStrategy()); + jcommander.addCommand("get-schema-compatibility-strategy", new GetSchemaCompatibilityStrategy()); } @Parameters(commandDescription = "Get max consumers per subscription for a topic") @@ -1684,6 +1688,53 @@ void run() throws PulsarAdminException { } } + + @Parameters(commandDescription = "Remove schema compatibility strategy on a topic") + private class RemoveSchemaCompatibilityStrategy extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + getAdmin().topicPolicies().removeSchemaCompatibilityStrategy(persistentTopic); + } + } + + @Parameters(commandDescription = "Set schema compatibility strategy on a topic") + private class SetSchemaCompatibilityStrategy extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = {"--strategy", "-s"}, description = "Schema compatibility strategy: [UNDEFINED, " + + "ALWAYS_INCOMPATIBLE, ALWAYS_COMPATIBLE, BACKWARD, FORWARD, FULL, BACKWARD_TRANSITIVE, " + + "FORWARD_TRANSITIVE, FULL_TRANSITIVE]", required = true) + private SchemaCompatibilityStrategy strategy; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + getAdmin().topicPolicies().setSchemaCompatibilityStrategy(persistentTopic, strategy); + } + } + + @Parameters(commandDescription = "Get schema compatibility strategy on a topic") + private class GetSchemaCompatibilityStrategy extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = {"-ap", "--applied"}, description = "Get the applied policy of the topic") + private boolean applied = false; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + SchemaCompatibilityStrategy strategy = + getAdmin().topicPolicies().getSchemaCompatibilityStrategy(persistentTopic, applied); + print(strategy == null ? "null" : strategy.name()); + } + } + private TopicPolicies getTopicPolicies(boolean isGlobal) { return getAdmin().topicPolicies(isGlobal); } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java index 8bca993edab5c..e925132678592 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java @@ -72,7 +72,6 @@ import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; 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.util.DateFormatter; import org.apache.pulsar.common.util.RelativeTimeUtil; @@ -245,10 +244,6 @@ public CmdTopics(Supplier admin) { jcommander.addCommand("set-replication-clusters", new SetReplicationClusters()); jcommander.addCommand("remove-replication-clusters", new RemoveReplicationClusters()); - jcommander.addCommand("remove-schema-compatibility-strategy", new RemoveSchemaCompatibilityStrategy()); - jcommander.addCommand("set-schema-compatibility-strategy", new SetSchemaCompatibilityStrategy()); - jcommander.addCommand("get-schema-compatibility-strategy", new GetSchemaCompatibilityStrategy()); - initDeprecatedCommands(); } @@ -2816,48 +2811,4 @@ void run() throws PulsarAdminException { } } - - @Parameters(commandDescription = "Remove schema compatibility strategy on a topic") - private class RemoveSchemaCompatibilityStrategy extends CliCommand { - @Parameter(description = "persistent://tenant/namespace/topic", required = true) - private java.util.List params; - - @Override - void run() throws PulsarAdminException { - String persistentTopic = validatePersistentTopic(params); - getAdmin().topicPolicies().removeSchemaCompatibilityStrategy(persistentTopic); - } - } - - @Parameters(commandDescription = "Set schema compatibility strategy on a topic") - private class SetSchemaCompatibilityStrategy extends CliCommand { - @Parameter(description = "persistent://tenant/namespace/topic", required = true) - private java.util.List params; - - @Parameter(names = {"--strategy", "-s"}, description = "Schema compatibility strategy: [UNDEFINED, " - + "ALWAYS_INCOMPATIBLE, ALWAYS_COMPATIBLE, BACKWARD, FORWARD, FULL, BACKWARD_TRANSITIVE, " - + "FORWARD_TRANSITIVE, FULL_TRANSITIVE]", required = true) - private SchemaCompatibilityStrategy strategy; - - @Override - void run() throws PulsarAdminException { - String persistentTopic = validatePersistentTopic(params); - getAdmin().topicPolicies().setSchemaCompatibilityStrategy(persistentTopic, strategy); - } - } - - @Parameters(commandDescription = "Get schema compatibility strategy on a topic") - private class GetSchemaCompatibilityStrategy extends CliCommand { - @Parameter(description = "persistent://tenant/namespace/topic", required = true) - private java.util.List params; - - @Parameter(names = {"-ap", "--applied"}, description = "Get the applied policy of the topic") - private boolean applied = false; - - @Override - void run() throws PulsarAdminException { - String persistentTopic = validatePersistentTopic(params); - print(getAdmin().topicPolicies().getSchemaCompatibilityStrategy(persistentTopic, applied)); - } - } } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/topicpolicies/SchemaCompatibilityStrategyTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/topicpolicies/SchemaCompatibilityStrategyTest.java index e82dfef43b79a..56d89d6b5028a 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/topicpolicies/SchemaCompatibilityStrategyTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/topicpolicies/SchemaCompatibilityStrategyTest.java @@ -19,67 +19,71 @@ package org.apache.pulsar.tests.integration.cli.topicpolicies; -import static org.testng.Assert.assertTrue; +import static org.testng.Assert.assertEquals; import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.tests.integration.docker.ContainerExecResult; -import org.apache.pulsar.tests.integration.suites.PulsarTestSuite; +import org.apache.pulsar.tests.integration.suites.PulsarCliTestSuite; import org.awaitility.Awaitility; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; -public class SchemaCompatibilityStrategyTest extends PulsarTestSuite { +public class SchemaCompatibilityStrategyTest extends PulsarCliTestSuite { + @BeforeClass(alwaysRun = true) @Override - public void setupCluster() throws Exception { - super.setupCluster(); + public void before() throws Exception { + enableTopicPolicies(); + super.before(); } + @BeforeClass(alwaysRun = true) @Override - public void tearDownCluster() throws Exception { - super.tearDownCluster(); + public void after() throws Exception { + super.after(); } @Test - public void testSchemaCompatibilityCmd() throws Exception { - String topicName = generateTopicName("",true); + public void testSchemaCompatibilityStrategyCmd() throws Exception { + String topicName = generateTopicName("test-schema-compatibility-strategy", true); pulsarAdmin.topics().createNonPartitionedTopic(topicName); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy " + topicName); - assertTrue(result.getStdout().contentEquals("UNDEFINED")); - }); + ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", + "get-schema-compatibility-strategy", topicName); + assertEquals(result.getStdout().trim(), "null"); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy --applied " + topicName); - assertTrue(result.getStdout().contentEquals("FULL")); - }); + result = pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", "get-schema-compatibility-strategy", + "--applied", topicName); + assertEquals(result.getStdout().trim(), SchemaCompatibilityStrategy.FULL.name()); pulsarAdmin.topicPolicies().removeSchemaCompatibilityStrategy(topicName); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy " + topicName); - assertTrue(result.getStdout().contentEquals("UNDEFINED")); + Awaitility.await().untilAsserted(() -> { + assertEquals(pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", + "get-schema-compatibility-strategy", topicName).getStdout().trim(), "null"); }); } @Test - public void testSchemaCompatibilityCmdWithNamespaceLevel() throws Exception { - String topicName = generateTopicName("",true); - pulsarAdmin.namespaces() - .setSchemaCompatibilityStrategy("public/default", SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE); + public void testSchemaCompatibilityStrategyCmdWithNamespaceLevel() throws Exception { + String ns = generateNamespaceName(); + String fullNS = "public/" + ns; + pulsarAdmin.namespaces().createNamespace("public/"+ns); + + String topicName = generateTopicName(ns, "test-schema-compatibility-strategy", + true); + pulsarAdmin.namespaces().setSchemaCompatibilityStrategy(fullNS, SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE); pulsarAdmin.topics().createNonPartitionedTopic(topicName); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy " + topicName); - assertTrue(result.getStdout().contentEquals("UNDEFINED")); - }); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy --applied " + topicName); - assertTrue(result.getStdout().contentEquals("ALWAYS_INCOMPATIBLE")); - }); + ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", + "get-schema-compatibility-strategy", topicName); + assertEquals(result.getStdout().trim(), "null"); + + result = pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", + "get-schema-compatibility-strategy", "--applied", topicName); + assertEquals(result.getStdout().trim(), SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE.name()); pulsarAdmin.namespaces() - .setSchemaCompatibilityStrategy("public/default", SchemaCompatibilityStrategy.UNDEFINED); - Awaitility.await().untilAsserted(()->{ - ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker("topics get-schema-compatibility-strategy " + topicName); - assertTrue(result.getStdout().contentEquals("UNDEFINED")); - }); + .setSchemaCompatibilityStrategy(fullNS, SchemaCompatibilityStrategy.UNDEFINED); + result = pulsarCluster.runAdminCommandOnAnyBroker("topicPolicies", "get-schema-compatibility-strategy", + topicName); + assertEquals(result.getStdout().trim(), "null"); } } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarCliTestSuite.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarCliTestSuite.java new file mode 100644 index 0000000000000..9921cf11e26ab --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarCliTestSuite.java @@ -0,0 +1,40 @@ +/** + * 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.tests.integration.suites; + +import org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; + +public abstract class PulsarCliTestSuite extends PulsarClusterTestBase { + @BeforeClass(alwaysRun = true) + public void before() throws Exception { + setup(); + } + + @AfterClass(alwaysRun = true) + public void after() throws Exception { + cleanup(); + } + + protected final void enableTopicPolicies() { + this.brokerEnvs.put("systemTopicEnabled", "true"); + this.brokerEnvs.put("topicLevelPoliciesEnabled", "true"); + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterTestBase.java index 0c153f0ca4ef3..34994f75216d4 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterTestBase.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterTestBase.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.tests.integration.topologies; +import java.util.HashMap; +import java.util.Map; import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.admin.PulsarAdmin; @@ -29,6 +31,8 @@ @Slf4j public abstract class PulsarClusterTestBase extends PulsarTestBase { + protected final Map brokerEnvs = new HashMap<>(); + @Override protected final void setup() throws Exception { setupCluster(); @@ -98,7 +102,8 @@ public void setupCluster(String namePrefix) throws Exception { .collect(joining("-")); PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder = PulsarClusterSpec.builder() - .clusterName(clusterName); + .clusterName(clusterName) + .brokerEnvs(brokerEnvs); setupCluster(beforeSetupCluster(clusterName, specBuilder).build()); } diff --git a/tests/integration/src/test/resources/pulsar-cli.xml b/tests/integration/src/test/resources/pulsar-cli.xml index 847f492d0322b..70b4136f2ca76 100644 --- a/tests/integration/src/test/resources/pulsar-cli.xml +++ b/tests/integration/src/test/resources/pulsar-cli.xml @@ -30,6 +30,7 @@ +