Allow specifying readerSchema on KafkaAvroDeserializer (#2366)

frankgrimes97 · rayokota · web-flow · commit 79a27211eb44 · 2022-08-29T11:28:51.000-07:00
* Allow specifying readerSchema on KafkaAvroDeserializer Adds two new configs: - `specific.avro.key.type` - `specific.avro.value.type` These configs are only applicable when `specific.avro.reader` is true. If unset, it defaults to current behavior of using the writerSchema. #670 * Minor change to logger name Co-authored-by: Robert Yokota <rayokota@gmail.com>
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializer.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializer.java
@@ -45,11 +45,13 @@
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import org.apache.kafka.common.config.ConfigException;
 
 public abstract class AbstractKafkaAvroDeserializer extends AbstractKafkaSchemaSerDe {
   private final DecoderFactory decoderFactory = DecoderFactory.get();
   protected boolean isKey;
   protected boolean useSpecificAvroReader = false;
+  protected Schema specificAvroReaderSchema = null;
   protected boolean avroReflectionAllowNull = false;
   protected boolean avroUseLogicalTypeConverters = false;
   private final Map<String, Schema> readerSchemaCache = new ConcurrentHashMap<>();
@@ -91,14 +93,34 @@ public DatumReader<?> load(IdentityPair<Schema, Schema> key) {
         .build(cacheLoader);
   }
 
+  protected void configure(KafkaAvroDeserializerConfig config) {
+    configure(config, null);
+  }
+
   /**
    * Sets properties for this deserializer without overriding the schema registry client itself.
    * Useful for testing, where a mock client is injected.
    */
-  protected void configure(KafkaAvroDeserializerConfig config) {
+  protected void configure(KafkaAvroDeserializerConfig config,  Class<?> type) {
     configureClientProperties(config, new AvroSchemaProvider());
     useSpecificAvroReader = config
         .getBoolean(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG);
+
+    if (useSpecificAvroReader && type != null) {
+      try {
+        specificAvroReaderSchema = ((SpecificRecord)type.getDeclaredConstructor().newInstance())
+                .getSchema();
+      } catch (Exception e) {
+        throw new ConfigException(
+          String.format(
+            "Error getting specificAvroReaderSchema from '%s'",
+            type.getName()
+          ),
+          e
+        );
+      }
+    }
+
     avroReflectionAllowNull = config
         .getBoolean(KafkaAvroDeserializerConfig.AVRO_REFLECTION_ALLOW_NULL_CONFIG);
     avroUseLogicalTypeConverters = config
@@ -123,7 +145,7 @@ protected KafkaAvroDeserializerConfig deserializerConfig(Properties props) {
    * @return the deserialized object
    */
   protected Object deserialize(byte[] payload) throws SerializationException {
-    return deserialize(null, isKey, payload, null);
+    return deserialize(null, isKey, payload, specificAvroReaderSchema);
   }
 
   /**
@@ -201,7 +223,7 @@ protected GenericContainerWithVersion deserializeWithSchemaAndVersion(
     // explicit from the Connector).
     DeserializationContext context = new DeserializationContext(topic, isKey, payload);
     AvroSchema schema = context.schemaForDeserialize();
-    Object result = context.read(schema.rawSchema(), null);
+    Object result = context.read(schema.rawSchema(), specificAvroReaderSchema);
 
     try {
       Integer version = schemaVersion(topic, isKey, context.getSchemaId(),
@@ -396,7 +418,7 @@ int getSchemaId() {
     }
 
     Object read(Schema writerSchema) {
-      return read(writerSchema, null);
+      return read(writerSchema, specificAvroReaderSchema);
     }
 
     Object read(Schema writerSchema, Schema readerSchema) {
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDecoder.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDecoder.java
@@ -29,15 +29,16 @@ public KafkaAvroDecoder(SchemaRegistryClient schemaRegistry) {
   }
 
   public KafkaAvroDecoder(SchemaRegistryClient schemaRegistry, VerifiableProperties props) {
+
     this.schemaRegistry = schemaRegistry;
-    configure(deserializerConfig(props.props()));
+    configure(deserializerConfig(props.props()), null);
   }
 
   /**
    * Constructor used by Kafka consumer.
    */
   public KafkaAvroDecoder(VerifiableProperties props) {
-    configure(new KafkaAvroDeserializerConfig(props.props()));
+    configure(new KafkaAvroDeserializerConfig(props.props()), null);
   }
 
   @Override
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializer.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializer.java
@@ -22,10 +22,19 @@
 import java.util.Map;
 
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
+import static io.confluent.kafka.serializers.KafkaAvroDeserializerConfig.SPECIFIC_AVRO_KEY_TYPE_CONFIG;
+import static io.confluent.kafka.serializers.KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG;
+import static io.confluent.kafka.serializers.KafkaAvroDeserializerConfig.SPECIFIC_AVRO_VALUE_TYPE_CONFIG;
+import org.apache.avro.specific.SpecificRecord;
+import org.apache.kafka.common.config.ConfigException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class KafkaAvroDeserializer extends AbstractKafkaAvroDeserializer
     implements Deserializer<Object> {
 
+  private static final Logger log = LoggerFactory.getLogger(KafkaAvroDeserializer.class);
+
   /**
    * Constructor used by Kafka consumer.
    */
@@ -38,19 +47,56 @@ public KafkaAvroDeserializer(SchemaRegistryClient client) {
   }
 
   public KafkaAvroDeserializer(SchemaRegistryClient client, Map<String, ?> props) {
-    schemaRegistry = client;
-    configure(deserializerConfig(props));
+    this(client, props, false);
   }
 
   @Override
-  public void configure(Map<String, ?> configs, boolean isKey) {
+  public void configure(Map<String, ?> props, boolean isKey) {
+    this.isKey = isKey;
+    configure(deserializerConfig(props), null);
+  }
+
+  public KafkaAvroDeserializer(SchemaRegistryClient client, Map<String, ?> props, boolean isKey) {
+    schemaRegistry = client;
     this.isKey = isKey;
-    configure(new KafkaAvroDeserializerConfig(configs));
+
+    final String specificAvroClassLookupKey = isKey
+         ? SPECIFIC_AVRO_KEY_TYPE_CONFIG :
+           SPECIFIC_AVRO_VALUE_TYPE_CONFIG;
+
+    final KafkaAvroDeserializerConfig config = deserializerConfig(props);
+
+    final Class<?> type = config.getClass(specificAvroClassLookupKey);
+
+    if (type != null && !config.getBoolean(SPECIFIC_AVRO_READER_CONFIG)) {
+      if (log.isWarnEnabled()) {
+        log.warn(
+            String.format(
+              "'%s' value of '%s' is ignored because '%s' is false",
+              specificAvroClassLookupKey,
+              type.getName(),
+              SPECIFIC_AVRO_READER_CONFIG
+            )
+        );
+      }
+    }
+
+    if (type != null && !SpecificRecord.class.isAssignableFrom(type)) {
+      throw new ConfigException(
+        String.format("Value '%s' specified for '%s' is not a '%s'",
+          type.getName(),
+          specificAvroClassLookupKey,
+          SpecificRecord.class.getName()
+        )
+      );
+    }
+
+    configure(deserializerConfig(props), type);
   }
 
   @Override
   public Object deserialize(String topic, byte[] bytes) {
-    return deserialize(topic, isKey, bytes, null);
+    return deserialize(topic, isKey, bytes, specificAvroReaderSchema);
   }
 
   /**
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializerConfig.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializerConfig.java
@@ -30,6 +30,18 @@ public class KafkaAvroDeserializerConfig extends AbstractKafkaSchemaSerDeConfig
   public static final String SPECIFIC_AVRO_READER_DOC =
       "If true, tries to look up the SpecificRecord class ";
 
+  public static final String SPECIFIC_AVRO_KEY_TYPE_CONFIG = "specific.avro.key.type";
+  public static final String SPECIFIC_AVRO_KEY_TYPE_DOC =
+      "A class generated by Avro that the message key should be deserialized to";
+
+  public static final String SPECIFIC_AVRO_VALUE_TYPE_CONFIG = "specific.avro.value.type";
+  public static final String SPECIFIC_AVRO_VALUE_TYPE_DOC =
+      "A class generated by Avro that the message value should be deserialized to";
+
+  public static final String AVRO_READER_SCHEMA_CONFIG = "avro.reader.schema";
+  public static final String AVRO_READER_SCHEMA_DOC =
+      "Reader schema to use when deserializing. Defaults to using the writer schema.";
+
   public static final String AVRO_REFLECTION_ALLOW_NULL_CONFIG = "avro.reflection.allow.null";
   public static final boolean AVRO_REFLECTION_ALLOW_NULL_DEFAULT = false;
   public static final String AVRO_REFLECTION_ALLOW_NULL_DOC =
@@ -47,6 +59,14 @@ public class KafkaAvroDeserializerConfig extends AbstractKafkaSchemaSerDeConfig
     config = baseConfigDef()
         .define(SPECIFIC_AVRO_READER_CONFIG, Type.BOOLEAN, SPECIFIC_AVRO_READER_DEFAULT,
             Importance.LOW, SPECIFIC_AVRO_READER_DOC)
+        .define(SPECIFIC_AVRO_KEY_TYPE_CONFIG,
+                Type.CLASS,
+                null,
+                Importance.MEDIUM, SPECIFIC_AVRO_KEY_TYPE_DOC)
+        .define(SPECIFIC_AVRO_VALUE_TYPE_CONFIG,
+                Type.CLASS,
+                null,
+                Importance.MEDIUM, SPECIFIC_AVRO_VALUE_TYPE_DOC)
         .define(AVRO_REFLECTION_ALLOW_NULL_CONFIG, Type.BOOLEAN, AVRO_REFLECTION_ALLOW_NULL_DEFAULT,
             Importance.MEDIUM, AVRO_REFLECTION_ALLOW_NULL_DOC)
         .define(AVRO_USE_LOGICAL_TYPE_CONVERTERS_CONFIG, ConfigDef.Type.BOOLEAN,
diff --git a/avro-serializer/src/test/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializerTest.java b/avro-serializer/src/test/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializerTest.java
@@ -6,7 +6,6 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 
-import io.confluent.kafka.schemaregistry.ParsedSchema;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
@@ -21,6 +20,7 @@
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
@@ -77,7 +77,7 @@ public void assertSchemaNotCopiedWhenDeserializedWithVersion(
     String subject = subjectNameStrategy.subjectName(topic, false,
         new AvroSchema(avroRecord.getSchema()));
     avroSerializer.configure(configs, false);
-    deserializer.configure(new KafkaAvroDeserializerConfig(configs));
+    deserializer.configure(new KafkaAvroDeserializerConfig(configs), null);
     schemaRegistry.register(subject, new AvroSchema(avroRecord.getSchema()));
     byte[] bytes = avroSerializer.serialize(topic, avroRecord);
     IndexedRecord deserialized
@@ -219,4 +219,56 @@ public void testMixedUrlsAreRejected() {
 
     Assert.assertNull(kafkaAvroSerializer.schemaRegistry);
   }
+
+  @Test
+  public void testInvalidSpecificSchemaKeyTypeConfig() {
+    HashMap<String, String> props = new HashMap<String, String>();
+    // Intentionally invalid schema registry URL to satisfy the config class's requirement that
+    // it be set.
+    props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "bogus");
+    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
+    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_VALUE_TYPE_CONFIG,
+        Object.class.getName()
+    );
+
+    try {
+      new KafkaAvroDeserializer(
+        new MockSchemaRegistryClient(),
+        props
+      );
+      fail();
+    }
+    catch (ConfigException e) {
+      Assert.assertEquals(
+        "Value 'java.lang.Object' specified for " +
+          "'specific.avro.value.type' is not a " +
+          "'org.apache.avro.specific.SpecificRecord'",
+        e.getMessage()
+      );
+    }
+
+
+    props.remove(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_VALUE_TYPE_CONFIG);
+    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_KEY_TYPE_CONFIG,
+        Object.class.getName()
+    );
+
+    try {
+      new KafkaAvroDeserializer(
+        new MockSchemaRegistryClient(),
+        props,
+        true
+      );
+      fail();
+    }
+    catch (ConfigException e) {
+      Assert.assertEquals(
+        "Value 'java.lang.Object' specified for " +
+          "'specific.avro.key.type' is not a " +
+          "'org.apache.avro.specific.SpecificRecord'",
+        e.getMessage()
+      );
+    }
+
+  }
 }
diff --git a/avro-serializer/src/test/java/io/confluent/kafka/serializers/KafkaAvroSerializerTest.java b/avro-serializer/src/test/java/io/confluent/kafka/serializers/KafkaAvroSerializerTest.java
@@ -58,7 +58,6 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -895,6 +894,32 @@ public void testKafkaAvroSerializerSpecificRecordWithProjection() {
     assertEquals("testUser", ((User) obj).getName().toString());
   }
 
+  @Test
+  public void testKafkaAvroSerializerSpecificRecordWithValueTypeConfig() {
+    HashMap<String, String> specificDeserializerProps = new HashMap<String, String>();
+    // Intentionally invalid schema registry URL to satisfy the config class's requirement that
+    // it be set.
+    specificDeserializerProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "bogus");
+    specificDeserializerProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
+    specificDeserializerProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_VALUE_TYPE_CONFIG,
+        User.class.getName()
+    );
+
+    final KafkaAvroDeserializer specificAvroDeserializerWithReaderSchema = new KafkaAvroDeserializer(
+      schemaRegistry, specificDeserializerProps
+    );
+
+    IndexedRecord avroRecord = createExtendedSpecificAvroRecord();
+    final byte[] bytes = avroSerializer.serialize(topic, avroRecord);
+
+    final Object obj = specificAvroDeserializerWithReaderSchema.deserialize(topic, bytes);
+    assertTrue(
+        "Full object should be a io.confluent.kafka.example.User",
+        User.class.isInstance(obj)
+    );
+    assertEquals("testUser", ((User) obj).getName().toString());
+  }
+
   @Test
   public void testKafkaAvroSerializerReflectionRecord() {
     byte[] bytes;

Original file line number	Diff line number	Diff line change
`@@ -29,15 +29,16 @@ public KafkaAvroDecoder(SchemaRegistryClient schemaRegistry) {`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`public KafkaAvroDecoder(SchemaRegistryClient schemaRegistry, VerifiableProperties props) {`
	`32`	`+`
`32`	`33`	`this.schemaRegistry = schemaRegistry;`
`33`		`- configure(deserializerConfig(props.props()));`
	`34`	`+ configure(deserializerConfig(props.props()), null);`
`34`	`35`	`}`
`35`	`36`
`36`	`37`	`/**`
`37`	`38`	`* Constructor used by Kafka consumer.`
`38`	`39`	`*/`
`39`	`40`	`public KafkaAvroDecoder(VerifiableProperties props) {`
`40`		`- configure(new KafkaAvroDeserializerConfig(props.props()));`
	`41`	`+ configure(new KafkaAvroDeserializerConfig(props.props()), null);`
`41`	`42`	`}`
`42`	`43`
`43`	`44`	`@Override`