You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2019/02/09 07:37:57 UTC
[pulsar] branch master updated: Added integration test for Python
and Java schema compatibility (#3557)
This is an automated email from the ASF dual-hosted git repository.
sijie 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 a1a20a8 Added integration test for Python and Java schema compatibility (#3557)
a1a20a8 is described below
commit a1a20a8836d5f54aecde2969d62295d78013eb36
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Feb 8 23:37:53 2019 -0800
Added integration test for Python and Java schema compatibility (#3557)
### Motivation
Added integration test to validate that a Java client with schema can interoperate with a Python client with equivalent schema declaration.
---
.../docker-images/latest-version-image/Dockerfile | 2 +
.../python-examples/consumer_schema.py | 49 ++++++++
.../python-examples/producer_schema.py | 45 ++++++++
.../tests/integration/python/PythonSchemaTest.java | 123 +++++++++++++++++++++
4 files changed, 219 insertions(+)
diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile
index 4ded790..ab7b7f8 100644
--- a/tests/docker-images/latest-version-image/Dockerfile
+++ b/tests/docker-images/latest-version-image/Dockerfile
@@ -46,3 +46,5 @@ RUN mkdir -p /pulsar/instances/deps
COPY python-examples/exclamation_lib.py /pulsar/instances/deps/
COPY python-examples/exclamation_with_extra_deps.py /pulsar/examples/python-examples/
COPY python-examples/exclamation.zip /pulsar/examples/python-examples/
+COPY python-examples/producer_schema.py /pulsar/examples/python-examples/
+COPY python-examples/consumer_schema.py /pulsar/examples/python-examples/
diff --git a/tests/docker-images/latest-version-image/python-examples/consumer_schema.py b/tests/docker-images/latest-version-image/python-examples/consumer_schema.py
new file mode 100755
index 0000000..64efcbe
--- /dev/null
+++ b/tests/docker-images/latest-version-image/python-examples/consumer_schema.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+import pulsar
+from pulsar.schema import *
+import sys
+
+
+class ReceiveExample(Record):
+ x = Integer()
+ y = Long()
+
+
+service_url = sys.argv[1]
+topic = sys.argv[2]
+
+client = pulsar.Client(service_url)
+
+consumer = client.subscribe(
+ topic=topic,
+ subscription_name="my-subscription",
+ schema=JsonSchema(ReceiveExample)
+ )
+
+msg = consumer.receive()
+
+obj = msg.value()
+
+assert obj.x == 1
+assert obj.y == 2
+
+client.close()
diff --git a/tests/docker-images/latest-version-image/python-examples/producer_schema.py b/tests/docker-images/latest-version-image/python-examples/producer_schema.py
new file mode 100755
index 0000000..bc85686
--- /dev/null
+++ b/tests/docker-images/latest-version-image/python-examples/producer_schema.py
@@ -0,0 +1,45 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+import pulsar
+from pulsar.schema import *
+import sys
+
+
+class MyExample(Record):
+ a = String()
+ b = Integer(required=True)
+
+
+service_url = sys.argv[1]
+topic = sys.argv[2]
+
+client = pulsar.Client(service_url)
+
+producer = client.create_producer(
+ topic=topic,
+ schema=AvroSchema(MyExample)
+ )
+
+producer.send(MyExample(a="Hello", b=1))
+
+client.close()
+
+sys.exit(0)
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/python/PythonSchemaTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/python/PythonSchemaTest.java
new file mode 100644
index 0000000..24b1f82
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/python/PythonSchemaTest.java
@@ -0,0 +1,123 @@
+/**
+ * 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.python;
+
+import static org.testng.Assert.assertEquals;
+
+import lombok.Cleanup;
+import lombok.Data;
+
+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.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
+import org.apache.pulsar.tests.integration.suites.PulsarTestSuite;
+import org.testng.annotations.Test;
+
+/**
+ * Test pulsar Python/Java schema interoperability
+ */
+public class PythonSchemaTest extends PulsarTestSuite {
+
+ @Data
+ static class Example1 {
+ private Integer x;
+ private Long y;
+ }
+
+ @Data
+ static class Example2 {
+ private String a;
+ private int b;
+ }
+
+ /**
+ * Publish from Java and consume from Python
+ */
+ @Test(dataProvider = "ServiceUrls")
+ public void testJavaPublishPythonConsume(String serviceUrl) throws Exception {
+ String nsName = generateNamespaceName();
+ pulsarCluster.createNamespace(nsName);
+
+ String topicName = generateTopicName(nsName, "testJavaPublishPythonConsume", true);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(serviceUrl)
+ .build();
+
+ // Create subscription to retain data
+ client.newConsumer(Schema.JSON(Example1.class))
+ .topic(topicName)
+ .subscriptionName("my-subscription")
+ .subscribe()
+ .close();
+
+ @Cleanup
+ Producer<Example1> producer = client.newProducer(Schema.JSON(Example1.class))
+ .topic(topicName)
+ .create();
+
+ Example1 e1 = new Example1();
+ e1.setX(1);
+ e1.setY(2L);
+ producer.send(e1);
+
+ // Verify Python can receive the typed message
+
+ ContainerExecResult res = pulsarCluster.getAnyBroker()
+ .execCmd("/pulsar/examples/python-examples/consumer_schema.py", "pulsar://localhost:6650", topicName);
+ assertEquals(res.getExitCode(), 0);
+ }
+
+ /**
+ * Publish from Java and consume from Python
+ */
+ @Test(dataProvider = "ServiceUrls")
+ public void testPythonPublishJavaConsume(String serviceUrl) throws Exception {
+ String nsName = generateNamespaceName();
+ pulsarCluster.createNamespace(nsName);
+
+ String topicName = generateTopicName(nsName, "testPythonPublishJavaConsume", true);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(serviceUrl)
+ .build();
+
+ @Cleanup
+ Consumer<Example2> consumer = client.newConsumer(Schema.AVRO(Example2.class))
+ .topic(topicName)
+ .subscriptionName("test-sub")
+ .subscribe();
+
+ // Verify Python can receive the typed message
+
+ ContainerExecResult res = pulsarCluster.getAnyBroker()
+ .execCmd("/pulsar/examples/python-examples/producer_schema.py", "pulsar://localhost:6650", topicName);
+ assertEquals(res.getExitCode(), 0);
+
+ Message<Example2> msg = consumer.receive();
+ Example2 e2 = msg.getValue();
+ assertEquals(e2.a, "Hello");
+ assertEquals(e2.b, 1);
+ }
+}
\ No newline at end of file