Fix serializer to use the right (Registry's) schema.

utils/serde/src/main/java/io/apicurio/registry/utils/serde/AbstractKafkaSerializer.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Serializer;
 
+import javax.ws.rs.core.Response;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -34,6 +35,8 @@
  */
 public abstract class AbstractKafkaSerializer<T, U, S extends AbstractKafkaSerializer<T, U, S>> extends AbstractKafkaStrategyAwareSerDe<T, S> implements Serializer<U> {
 
+    private SchemaCache<T> cache;
+
     public AbstractKafkaSerializer() {
         this(null);
     }
@@ -50,6 +53,20 @@ public AbstractKafkaSerializer(
         super(client, artifactIdStrategy, globalIdStrategy);
     }
 
+    public synchronized SchemaCache<T> getCache() {
+        if (cache == null) {
+            cache = new SchemaCache<T>(getClient()) {
+                @Override
+                protected T toSchema(Response response) {
+                    return readSchema(response);
+                }
+            };
+        }
+        return cache;
+    }
+
+    protected abstract T readSchema(Response response);
+
     protected abstract T toSchema(U data);
 
     protected abstract ArtifactType artifactType();
@@ -73,6 +90,7 @@ public byte[] serialize(String topic, Headers headers, U data) {
             T schema = toSchema(data);
             String artifactId = getArtifactIdStrategy().artifactId(topic, isKey(), schema);
             long id = getGlobalIdStrategy().findId(getClient(), artifactId, artifactType(), schema);
+            schema = getCache().getSchema(id); // use registry's schema!
             ByteArrayOutputStream out = new ByteArrayOutputStream();
             if (headerUtils != null) {
                 headerUtils.addSchemaHeaders(headers, artifactId, id);

utils/serde/src/main/java/io/apicurio/registry/utils/serde/AvroKafkaDeserializer.java

@@ -18,11 +18,10 @@
 package io.apicurio.registry.utils.serde;
 
 import io.apicurio.registry.client.RegistryService;
-import io.apicurio.registry.utils.IoUtil;
 import io.apicurio.registry.utils.serde.avro.AvroDatumProvider;
-import io.apicurio.registry.utils.serde.avro.AvroSchemaUtils;
 import io.apicurio.registry.utils.serde.avro.DefaultAvroDatumProvider;
 import io.apicurio.registry.utils.serde.util.HeaderUtils;
+import io.apicurio.registry.utils.serde.util.ResponseUtils;
 import org.apache.avro.Schema;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DecoderFactory;
@@ -31,7 +30,6 @@
 import javax.ws.rs.core.Response;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.Map;
@@ -79,12 +77,7 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
     @Override
     protected Schema toSchema(Response response) {
-        Object responseEntity = response.getEntity();
-        if (responseEntity instanceof InputStream) {
-            return AvroSchemaUtils.parse(IoUtil.toString((InputStream) responseEntity));
-        } else {
-            return AvroSchemaUtils.parse(response.readEntity(String.class));
-        }
+        return ResponseUtils.toAvroSchema(response);
     }
 
     @Override

utils/serde/src/main/java/io/apicurio/registry/utils/serde/AvroKafkaSerializer.java

@@ -25,13 +25,15 @@
 import io.apicurio.registry.utils.serde.strategy.ArtifactIdStrategy;
 import io.apicurio.registry.utils.serde.strategy.GlobalIdStrategy;
 import io.apicurio.registry.utils.serde.util.HeaderUtils;
+import io.apicurio.registry.utils.serde.util.ResponseUtils;
 import io.apicurio.registry.utils.serde.util.Utils;
 import org.apache.avro.Schema;
 import org.apache.avro.io.DatumWriter;
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.EncoderFactory;
 import org.apache.kafka.common.header.Headers;
 
+import javax.ws.rs.core.Response;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -87,6 +89,11 @@ public void configure(Map<String, ?> configs, boolean isKey) {
         avroDatumProvider.configure(configs);
     }
 
+    @Override
+    protected Schema readSchema(Response response) {
+        return ResponseUtils.toAvroSchema(response);
+    }
+
     @Override
     protected Schema toSchema(U data) {
         return avroDatumProvider.toSchema(data);

utils/serde/src/main/java/io/apicurio/registry/utils/serde/ProtobufKafkaDeserializer.java

@@ -20,16 +20,15 @@
 import com.google.protobuf.DynamicMessage;
 import io.apicurio.registry.client.RegistryService;
 import io.apicurio.registry.common.proto.Serde;
-import io.apicurio.registry.utils.IoUtil;
+import io.apicurio.registry.utils.serde.util.ResponseUtils;
 import org.apache.kafka.common.header.Headers;
 
+import javax.ws.rs.core.Response;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import javax.ws.rs.core.Response;
 
 /**
  * @author Ales Justin
@@ -45,12 +44,7 @@ public ProtobufKafkaDeserializer(RegistryService client) {
 
     @Override
     protected byte[] toSchema(Response response) {
-        Object responseEntity = response.getEntity();
-        if (responseEntity instanceof InputStream) {
-            return IoUtil.toBytes((InputStream) responseEntity);
-        } else {
-            return response.readEntity(byte[].class);
-        }
+        return ResponseUtils.toBytesSchema(response);
     }
 
     @Override

utils/serde/src/main/java/io/apicurio/registry/utils/serde/ProtobufKafkaSerializer.java

@@ -23,8 +23,10 @@
 import io.apicurio.registry.types.ArtifactType;
 import io.apicurio.registry.utils.serde.strategy.ArtifactIdStrategy;
 import io.apicurio.registry.utils.serde.strategy.GlobalIdStrategy;
+import io.apicurio.registry.utils.serde.util.ResponseUtils;
 import org.apache.kafka.common.header.Headers;
 
+import javax.ws.rs.core.Response;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -45,6 +47,11 @@ public ProtobufKafkaSerializer(RegistryService client, ArtifactIdStrategy<byte[]
         super(client, artifactIdStrategy, idStrategy);
     }
 
+    @Override
+    protected byte[] readSchema(Response response) {
+        return ResponseUtils.toBytesSchema(response);
+    }
+
     @Override
     protected byte[] toSchema(U data) {
         Serde.Schema schema = toSchemaProto(data.getDescriptorForType().getFile());

utils/serde/src/main/java/io/apicurio/registry/utils/serde/util/ResponseUtils.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2021 Red Hat
+ *
+ * Licensed 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 io.apicurio.registry.utils.serde.util;
+
+import io.apicurio.registry.utils.IoUtil;
+import io.apicurio.registry.utils.serde.avro.AvroSchemaUtils;
+import org.apache.avro.Schema;
+
+import javax.ws.rs.core.Response;
+import java.io.InputStream;
+
+/**
+ * @author Ales Justin
+ */
+public class ResponseUtils {
+    public static byte[] toBytesSchema(Response response) {
+        Object responseEntity = response.getEntity();
+        if (responseEntity instanceof InputStream) {
+            return IoUtil.toBytes((InputStream) responseEntity);
+        } else {
+            return response.readEntity(byte[].class);
+        }
+    }
+
+    public static Schema toAvroSchema(Response response) {
+        Object responseEntity = response.getEntity();
+        if (responseEntity instanceof InputStream) {
+            return AvroSchemaUtils.parse(IoUtil.toString((InputStream) responseEntity));
+        } else {
+            return AvroSchemaUtils.parse(response.readEntity(String.class));
+        }
+    }
+
+}