You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by iv...@apache.org on 2018/10/15 09:58:18 UTC
[pulsar] branch master updated: CLI for setting schema update
policy (#2788)
This is an automated email from the ASF dual-hosted git repository.
ivank pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b8423a5 CLI for setting schema update policy (#2788)
b8423a5 is described below
commit b8423a5efbcb049b94dbdc350dc5d3159d9fdfbf
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Mon Oct 15 11:58:12 2018 +0200
CLI for setting schema update policy (#2788)
Allows administrator of a namespace to specify how new schemas
provided by producers are validated. The policies are backward,
forward, full and disabled. If disabled, the administrator can still
update the schema via the schemas admin tool.
---
.../org/apache/pulsar/admin/cli/CmdNamespaces.java | 50 +++
.../integration/cli/SchemaUpdateStrategyTest.java | 341 +++++++++++++++++++++
.../integration/containers/ChaosContainer.java | 7 +
3 files changed, 398 insertions(+)
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 8a86aac..cc1654b 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
@@ -41,6 +41,7 @@ import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
@Parameters(commandDescription = "Operations about namespaces")
@@ -880,6 +881,53 @@ public class CmdNamespaces extends CmdBase {
}
}
+ @Parameters(commandDescription = "Get the schema auto-update strategy for a namespace")
+ private class GetSchemaAutoUpdateStrategy extends CliCommand {
+ @Parameter(description = "tenant/namespace", required = true)
+ private java.util.List<String> params;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String namespace = validateNamespace(params);
+ System.out.println(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace)
+ .toString().toUpperCase());
+ }
+ }
+
+ @Parameters(commandDescription = "Set the schema auto-update strategy for a namespace")
+ private class SetSchemaAutoUpdateStrategy extends CliCommand {
+ @Parameter(description = "tenant/namespace", required = true)
+ private java.util.List<String> params;
+
+ @Parameter(names = { "--compatibility", "-c" },
+ description = "Compatibility level required for new schemas created via a Producer. "
+ + "Possible values (Full, Backward, Forward).")
+ private String strategyParam = null;
+
+ @Parameter(names = { "--disabled" }, description = "Disable automatic schema updates")
+ private boolean disabled = false;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String namespace = validateNamespace(params);
+
+ SchemaAutoUpdateCompatibilityStrategy strategy = null;
+ String strategyStr = strategyParam != null ? strategyParam.toUpperCase() : "";
+ if (disabled) {
+ strategy = SchemaAutoUpdateCompatibilityStrategy.AutoUpdateDisabled;
+ } else if (strategyStr.equals("FULL")) {
+ strategy = SchemaAutoUpdateCompatibilityStrategy.Full;
+ } else if (strategyStr.equals("BACKWARD")) {
+ strategy = SchemaAutoUpdateCompatibilityStrategy.Backward;
+ } else if (strategyStr.equals("FORWARD")) {
+ strategy = SchemaAutoUpdateCompatibilityStrategy.Forward;
+ } else {
+ throw new PulsarAdminException("Either --compatibility or --disabled must be specified");
+ }
+ admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(namespace, strategy);
+ }
+ }
+
public CmdNamespaces(PulsarAdmin admin) {
super("namespaces", admin);
jcommander.addCommand("list", new GetNamespacesPerProperty());
@@ -949,5 +997,7 @@ public class CmdNamespaces extends CmdBase {
jcommander.addCommand("set-offload-deletion-lag", new SetOffloadDeletionLag());
jcommander.addCommand("clear-offload-deletion-lag", new ClearOffloadDeletionLag());
+ jcommander.addCommand("get-schema-autoupdate-strategy", new GetSchemaAutoUpdateStrategy());
+ jcommander.addCommand("set-schema-autoupdate-strategy", new SetSchemaAutoUpdateStrategy());
}
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/SchemaUpdateStrategyTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/SchemaUpdateStrategyTest.java
new file mode 100644
index 0000000..c8c4692
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/SchemaUpdateStrategyTest.java
@@ -0,0 +1,341 @@
+/**
+ * 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.cli;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.reflect.AvroAlias;
+import org.apache.avro.reflect.AvroDefault;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
+import org.apache.pulsar.tests.integration.suites.PulsarTestSuite;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test setting the schema update strategy via the CLI.
+ */
+public class SchemaUpdateStrategyTest extends PulsarTestSuite {
+ private final static Logger log = LoggerFactory.getLogger(SchemaUpdateStrategyTest.class);
+
+ private void testAutoUpdateBackward(String namespace, String topicName) throws Exception {
+ ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "get-schema-autoupdate-strategy", namespace);
+ Assert.assertEquals(result.getStdout().trim(), "FULL");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "set-schema-autoupdate-strategy",
+ "--compatibility", "BACKWARD", namespace);
+
+ try (PulsarClient pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl()).build()) {
+ try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topicName).create()) {
+ p.send(new V1Data("test1", 1));
+ }
+
+ log.info("try with forward compat, should fail");
+ try (Producer<V3Data> p = pulsarClient.newProducer(Schema.AVRO(V3Data.class)).topic(topicName).create()) {
+ Assert.fail("Forward compat schema should be rejected");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with backward compat, should succeed");
+ try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class)).topic(topicName).create()) {
+ p.send(new V2Data("test2"));
+ }
+ }
+ }
+
+ private void testAutoUpdateForward(String namespace, String topicName) throws Exception {
+ ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "get-schema-autoupdate-strategy", namespace);
+ Assert.assertEquals(result.getStdout().trim(), "FULL");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "set-schema-autoupdate-strategy",
+ "--compatibility", "FORWARD", namespace);
+
+ try (PulsarClient pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl()).build()) {
+
+ try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topicName).create()) {
+ p.send(new V1Data("test1", 1));
+ }
+
+ log.info("try with backward compat, should fail");
+ try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class)).topic(topicName).create()) {
+ Assert.fail("Backward compat schema should be rejected");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with forward compat, should succeed");
+ try (Producer<V3Data> p = pulsarClient.newProducer(Schema.AVRO(V3Data.class)).topic(topicName).create()) {
+ p.send(new V3Data("test2", 1, 2));
+ }
+ }
+
+ }
+
+ private void testAutoUpdateFull(String namespace, String topicName) throws Exception {
+ ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "get-schema-autoupdate-strategy", namespace);
+ Assert.assertEquals(result.getStdout().trim(), "FULL");
+
+ try (PulsarClient pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl()).build()) {
+ try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topicName).create()) {
+ p.send(new V1Data("test1", 1));
+ }
+
+ log.info("try with backward compat only, should fail");
+ try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class)).topic(topicName).create()) {
+ Assert.fail("Backward compat only schema should fail");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with forward compat only, should fail");
+ try (Producer<V3Data> p = pulsarClient.newProducer(Schema.AVRO(V3Data.class)).topic(topicName).create()) {
+ Assert.fail("Forward compat only schema should fail");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with fully compat");
+ try (Producer<V4Data> p = pulsarClient.newProducer(Schema.AVRO(V4Data.class)).topic(topicName).create()) {
+ p.send(new V4Data("test2", 1, (short)100));
+ }
+ }
+ }
+
+ private void testAutoUpdateDisabled(String namespace, String topicName) throws Exception {
+ ContainerExecResult result = pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "get-schema-autoupdate-strategy", namespace);
+ Assert.assertEquals(result.getStdout().trim(), "FULL");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "set-schema-autoupdate-strategy",
+ "--disabled", namespace);
+
+ try (PulsarClient pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl()).build()) {
+ try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topicName).create()) {
+ p.send(new V1Data("test1", 1));
+ }
+ log.info("try with backward compat only, should fail");
+ try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class)).topic(topicName).create()) {
+ Assert.fail("Backward compat only schema should fail");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with forward compat only, should fail");
+ try (Producer<V3Data> p = pulsarClient.newProducer(Schema.AVRO(V3Data.class)).topic(topicName).create()) {
+ Assert.fail("Forward compat only schema should fail");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("try with fully compat, should fail");
+ try (Producer<V4Data> p = pulsarClient.newProducer(Schema.AVRO(V4Data.class)).topic(topicName).create()) {
+ Assert.fail("Fully compat schema should fail, autoupdate disabled");
+ } catch (PulsarClientException e) {
+ Assert.assertTrue(e.getMessage().contains("IncompatibleSchemaException"));
+ }
+
+ log.info("Manually set new schema");
+ ObjectMapper mapper = new ObjectMapper();
+ Map<String, String> schema = new HashMap<>();
+ schema.put("type", "AVRO");
+ schema.put("schema", new String(Schema.AVRO(V4Data.class).getSchemaInfo().getSchema(), UTF_8));
+ BrokerContainer b = pulsarCluster.getAnyBroker();
+ String schemaFile = String.format("/tmp/schema-%s", UUID.randomUUID().toString());
+ b.putFile(schemaFile, mapper.writeValueAsBytes(schema));
+
+ b.execCmd(PulsarCluster.ADMIN_SCRIPT, "schemas", "upload", "-f", schemaFile, topicName);
+
+ boolean success = false;
+ for (int i = 0; i < 50; i++) {
+ try (Producer<V4Data> p = pulsarClient.newProducer(Schema.AVRO(V4Data.class))
+ .topic(topicName).create()) {
+ p.send(new V4Data("test2", 1, (short)100));
+ success = true;
+ break;
+ } catch (Throwable t) {
+ // expected a few times until the broker sees the new schema
+ }
+ Thread.sleep(100);
+ }
+ Assert.assertTrue(success, "Should have been able to use new schema");
+ }
+ }
+
+ @AvroAlias(space="blah", alias="data")
+ static class V1Data {
+ String foo;
+ int bar;
+
+ V1Data(String foo, int bar) {
+ this.foo = foo;
+ this.bar = bar;
+ }
+ }
+
+ // backward compatible with V1Data
+ @AvroAlias(space="blah", alias="data")
+ static class V2Data {
+ String foo;
+
+ V2Data(String foo) {
+ this.foo = foo;
+ }
+ }
+
+ // forward compatible with V1Data
+ @AvroAlias(space="blah", alias="data")
+ static class V3Data {
+ String foo;
+ int bar;
+ long baz;
+
+ V3Data(String foo, int bar, long baz) {
+ this.foo = foo;
+ this.bar = bar;
+ this.baz = baz;
+ }
+ }
+
+ // fully compatible with V1Data
+ @AvroAlias(space="blah", alias="data")
+ static class V4Data {
+ String foo;
+ int bar;
+ @AvroDefault(value = "10")
+ short blah;
+
+ V4Data(String foo, int bar, short blah) {
+ this.foo = foo;
+ this.bar = bar;
+ this.blah = blah;
+ }
+ }
+
+ @Test
+ public void testBackwardV2() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/bw-p-v2");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/bw-np-v2");
+
+ testAutoUpdateBackward("public/bw-p-v2", "persistent://public/bw-p-v2/topic1");
+ testAutoUpdateBackward("public/bw-np-v2", "non-persistent://public/bw-np-v2/topic1");
+ }
+
+ @Test
+ public void testForwardV2() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/fw-p-v2");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/fw-np-v2");
+
+ testAutoUpdateForward("public/fw-p-v2", "persistent://public/fw-p-v2/topic1");
+ testAutoUpdateForward("public/fw-np-v2", "non-persistent://public/fw-np-v2/topic1");
+ }
+
+ @Test
+ public void testFullV2() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/full-p-v2");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/full-np-v2");
+
+ testAutoUpdateFull("public/full-p-v2", "persistent://public/full-p-v2/topic1");
+ testAutoUpdateFull("public/full-np-v2", "non-persistent://public/full-np-v2/topic1");
+ }
+
+ @Test
+ public void testDisabledV2() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/dis-p-v2");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create", "-c",
+ pulsarCluster.getClusterName(), "public/dis-np-v2");
+
+ testAutoUpdateDisabled("public/dis-p-v2", "persistent://public/dis-p-v2/topic1");
+ testAutoUpdateDisabled("public/dis-np-v2", "non-persistent://public/dis-np-v2/topic1");
+ }
+
+ @Test
+ public void testBackwardV1() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/b-p-v1");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/b-np-v1");
+ testAutoUpdateBackward("public/" + pulsarCluster.getClusterName() + "/b-p-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/b-p-v1/topic1");
+ testAutoUpdateBackward("public/" + pulsarCluster.getClusterName() + "/b-np-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/b-np-v1/topic1");
+ }
+
+ @Test
+ public void testForwardV1() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/f-p-v1");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/f-np-v1");
+ testAutoUpdateForward("public/" + pulsarCluster.getClusterName() + "/f-p-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/f-p-v1/topic1");
+ testAutoUpdateForward("public/" + pulsarCluster.getClusterName() + "/f-np-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/f-np-v1/topic1");
+ }
+
+ @Test
+ public void testFullV1() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/full-p-v1");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/full-np-v1");
+ testAutoUpdateFull("public/" + pulsarCluster.getClusterName() + "/full-p-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/full-p-v1/topic1");
+ testAutoUpdateFull("public/" + pulsarCluster.getClusterName() + "/full-np-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/full-np-v1/topic1");
+ }
+
+ @Test
+ public void testDisabledV1() throws Exception {
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/dis-p-v1");
+ pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "create",
+ "public/" + pulsarCluster.getClusterName() + "/dis-np-v1");
+ testAutoUpdateDisabled("public/" + pulsarCluster.getClusterName() + "/dis-p-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/dis-p-v1/topic1");
+ testAutoUpdateDisabled("public/" + pulsarCluster.getClusterName() + "/dis-np-v1",
+ "persistent://public/" + pulsarCluster.getClusterName() + "/dis-np-v1/topic1");
+ }
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java
index b211b5b..18ba9fb 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java
@@ -24,6 +24,7 @@ import com.github.dockerjava.api.DockerClient;
import com.github.dockerjava.api.command.LogContainerCmd;
import com.github.dockerjava.api.model.Frame;
import com.github.dockerjava.core.command.LogContainerResultCallback;
+import java.util.Base64;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -103,6 +104,12 @@ public class ChaosContainer<SelfT extends ChaosContainer<SelfT>> extends Generic
return sb.toString();
}
+ public void putFile(String path, byte[] contents) throws Exception {
+ String base64contents = Base64.getEncoder().encodeToString(contents);
+ String cmd = String.format("echo %s | base64 -d > %s", base64contents, path);
+ execCmd("bash", "-c", cmd);
+ }
+
public ContainerExecResult execCmd(String... commands) throws Exception {
DockerClient client = this.getDockerClient();
String dockerId = this.getContainerId();