schema_registry/test: Test avro and proto serializer compatibility

Signed-off-by: Ben Pope <ben@vectorized.io>

tests/rptest/tests/schema_registry_test.py

@@ -20,6 +20,7 @@
 
 from rptest.clients.types import TopicSpec
 from rptest.clients.kafka_cli_tools import KafkaCliTools
+from rptest.clients.python_librdkafka_serde_client import SerdeClient, SchemaType
 from rptest.tests.redpanda_test import RedpandaTest
 from rptest.services.redpanda import ResourceSettings
 
@@ -1013,3 +1014,29 @@ def test_protobuf(self):
         self.logger.info(result_raw)
         assert result_raw.status_code == requests.codes.ok
         assert result_raw.json() == [2]
+
+    @cluster(num_nodes=3)
+    def test_serde_client(self):
+        """
+        Verify basic serialization client
+        """
+        protocols = [SchemaType.AVRO, SchemaType.PROTOBUF]
+        topics = [f"serde-topic-{x.name}" for x in protocols]
+        self._create_topics(topics)
+        schema_reg = self.redpanda.schema_reg().split(',', 1)[0]
+        for i in range(len(protocols)):
+            self.logger.info(
+                f"Connecting to redpanda: {self.redpanda.brokers()} schema_reg: {schema_reg}"
+            )
+            client = SerdeClient(self.redpanda.brokers(),
+                                 schema_reg,
+                                 protocols[i],
+                                 topic=topics[i],
+                                 logger=self.logger)
+            client.run(2)
+            schema = self._get_subjects_subject_versions_version(f"{topics[i]}-value","latest")
+            self.logger.info(schema.json())
+            if protocols[i] == SchemaType.AVRO:
+                assert schema.json().get("schemaType") is None
+            else:
+                assert schema.json()["schemaType"] == protocols[i].name