Rename to SchemaRegistryApacheAvroSerializer. Group name is optional. (Azure#24622)

* Rename parameter from schemaFormat -> format.

* Updating schemaId -> Id.

* Updating schemaId -> Id.

* Update sample to mention getting properties.

* Rename to SchemaRegistryApacheAvroSerializer.

* Move optional parameters to options bag.

* Renaming to ApacheAvroSerializerBuilder

* Using Context.NONE if context is not passed in.

Co-authored-by: Srikanta <51379715+srnagar@users.noreply.github.com>

Author: conniey

sdk/schemaregistry/azure-data-schemaregistry-avro/src/main/java/com/azure/data/schemaregistry/avro/SchemaRegistryAvroSerializer.java renamed to sdk/schemaregistry/azure-data-schemaregistry-avro/src/main/java/com/azure/data/schemaregistry/avro/SchemaRegistryApacheAvroSerializer.java

@@ -24,28 +24,33 @@
 import static com.azure.core.util.FluxUtil.monoError;
 
 /**
- * Schema Registry-based serializer implementation for Avro data format.
+ * Schema Registry-based serializer implementation for Avro data format using Apache Avro.
  */
-public final class SchemaRegistryAvroSerializer implements ObjectSerializer {
+public final class SchemaRegistryApacheAvroSerializer implements ObjectSerializer {
 
     static final byte[] RECORD_FORMAT_INDICATOR = new byte[]{0x00, 0x00, 0x00, 0x00};
     static final int SCHEMA_ID_SIZE = 32;
     static final int RECORD_FORMAT_INDICATOR_SIZE = 4;
 
-    private final ClientLogger logger = new ClientLogger(SchemaRegistryAvroSerializer.class);
+    private final ClientLogger logger = new ClientLogger(SchemaRegistryApacheAvroSerializer.class);
     private final SchemaRegistryAsyncClient schemaRegistryClient;
     private final AvroSerializer avroSerializer;
-    private final String schemaGroup;
-    private final boolean autoRegisterSchemas;
+    private final SerializerOptions serializerOptions;
 
-    SchemaRegistryAvroSerializer(SchemaRegistryAsyncClient schemaRegistryClient,
-        AvroSerializer avroSerializer, String schemaGroup, boolean autoRegisterSchemas) {
+    /**
+     * Creates a new instance.
+     *
+     * @param schemaRegistryClient Client that interacts with Schema Registry.
+     * @param avroSerializer Serializer implemented using Apache Avro.
+     * @param serializerOptions Options to configure the serializer with.
+     */
+    SchemaRegistryApacheAvroSerializer(SchemaRegistryAsyncClient schemaRegistryClient,
+        AvroSerializer avroSerializer, SerializerOptions serializerOptions) {
         this.schemaRegistryClient = Objects.requireNonNull(schemaRegistryClient,
             "'schemaRegistryClient' cannot be null.");
         this.avroSerializer = Objects.requireNonNull(avroSerializer,
             "'avroSchemaRegistryUtils' cannot be null.");
-        this.schemaGroup = Objects.requireNonNull(schemaGroup, "'schemaGroup' cannot be null.");
-        this.autoRegisterSchemas = autoRegisterSchemas;
+        this.serializerOptions = Objects.requireNonNull(serializerOptions, "'serializerOptions' cannot be null.");
     }
 
     /**
@@ -166,7 +171,7 @@ public Mono<Void> serializeAsync(OutputStream outputStream, Object object) {
             return monoError(logger, exception);
         }
 
-        return this.maybeRegisterSchema(this.schemaGroup, schema.getFullName(), schema.toString())
+        return this.maybeRegisterSchema(serializerOptions.getSchemaGroup(), schema.getFullName(), schema.toString())
             .handle((id, sink) -> {
                 ByteBuffer recordFormatIndicatorBuffer = ByteBuffer
                     .allocate(RECORD_FORMAT_INDICATOR_SIZE)
@@ -196,13 +201,13 @@ public Mono<Void> serializeAsync(OutputStream outputStream, Object object) {
      * @return string representation of schema ID
      */
     private Mono<String> maybeRegisterSchema(String schemaGroup, String schemaName, String schemaString) {
-        if (this.autoRegisterSchemas) {
+        if (serializerOptions.autoRegisterSchemas()) {
             return this.schemaRegistryClient
                 .registerSchema(schemaGroup, schemaName, schemaString, SchemaFormat.AVRO)
-                .map(SchemaProperties::getSchemaId);
+                .map(SchemaProperties::getId);
         } else {
             return this.schemaRegistryClient.getSchemaProperties(
-                schemaGroup, schemaName, schemaString, SchemaFormat.AVRO).map(properties -> properties.getSchemaId());
+                schemaGroup, schemaName, schemaString, SchemaFormat.AVRO).map(properties -> properties.getId());
         }
     }
 

sdk/schemaregistry/azure-data-schemaregistry-avro/src/main/java/com/azure/data/schemaregistry/avro/SchemaRegistryAvroSerializerBuilder.java renamed to sdk/schemaregistry/azure-data-schemaregistry-avro/src/main/java/com/azure/data/schemaregistry/avro/SchemaRegistryApacheAvroSerializerBuilder.java

@@ -11,12 +11,13 @@
 import org.apache.avro.specific.SpecificRecord;
 
 /**
- * The builder implementation for building {@link SchemaRegistryAvroSerializer}.
+ * The builder implementation for building {@link SchemaRegistryApacheAvroSerializer}.
  *
- * @see SchemaRegistryAvroSerializer
+ * @see SchemaRegistryApacheAvroSerializer
  */
-public final class SchemaRegistryAvroSerializerBuilder {
+public final class SchemaRegistryApacheAvroSerializerBuilder {
     private static final boolean AVRO_SPECIFIC_READER_DEFAULT = false;
+    private static final int MAX_CACHE_SIZE = 128;
 
     private Boolean autoRegisterSchemas;
     private Boolean avroSpecificReader;
@@ -26,22 +27,23 @@ public final class SchemaRegistryAvroSerializerBuilder {
     /**
      * Instantiates instance of Builder class. Supplies client defaults.
      */
-    public SchemaRegistryAvroSerializerBuilder() {
+    public SchemaRegistryApacheAvroSerializerBuilder() {
         this.autoRegisterSchemas = false;
         this.avroSpecificReader = false;
     }
 
     /**
-     * Specifies schema group for interacting with Azure Schema Registry service.
+     * Specifies schema group for interacting with Azure Schema Registry service. This is optional unless
+     * {@link #autoRegisterSchema(boolean) autoRegisterSchema} is set to {@code true}.
      *
      * If auto-registering schemas, schema will be stored under this group. If not auto-registering, serializer will
      * request schema ID for matching data schema under specified group.
      *
      * @param schemaGroup Azure Schema Registry schema group
      *
-     * @return updated {@link SchemaRegistryAvroSerializerBuilder} instance
+     * @return updated {@link SchemaRegistryApacheAvroSerializerBuilder} instance
      */
-    public SchemaRegistryAvroSerializerBuilder schemaGroup(String schemaGroup) {
+    public SchemaRegistryApacheAvroSerializerBuilder schemaGroup(String schemaGroup) {
         this.schemaGroup = schemaGroup;
         return this;
     }
@@ -57,9 +59,9 @@ public SchemaRegistryAvroSerializerBuilder schemaGroup(String schemaGroup) {
      *
      * @param autoRegisterSchemas flag for schema auto-registration
      *
-     * @return updated {@link SchemaRegistryAvroSerializerBuilder} instance
+     * @return updated {@link SchemaRegistryApacheAvroSerializerBuilder} instance
      */
-    public SchemaRegistryAvroSerializerBuilder autoRegisterSchema(boolean autoRegisterSchemas) {
+    public SchemaRegistryApacheAvroSerializerBuilder autoRegisterSchema(boolean autoRegisterSchemas) {
         this.autoRegisterSchemas = autoRegisterSchemas;
         return this;
     }
@@ -71,9 +73,9 @@ public SchemaRegistryAvroSerializerBuilder autoRegisterSchema(boolean autoRegist
      * @param avroSpecificReader {@code true} to deserialize into {@link SpecificRecord} via {@link
      *     SpecificDatumReader}; {@code false} otherwise.
      *
-     * @return updated {@link SchemaRegistryAvroSerializerBuilder} instance.
+     * @return updated {@link SchemaRegistryApacheAvroSerializerBuilder} instance.
      */
-    public SchemaRegistryAvroSerializerBuilder avroSpecificReader(boolean avroSpecificReader) {
+    public SchemaRegistryApacheAvroSerializerBuilder avroSpecificReader(boolean avroSpecificReader) {
         this.avroSpecificReader = avroSpecificReader;
         return this;
     }
@@ -83,31 +85,32 @@ public SchemaRegistryAvroSerializerBuilder avroSpecificReader(boolean avroSpecif
      *
      * @param schemaRegistryAsyncClient The {@link SchemaRegistryAsyncClient}.
      *
-     * @return updated {@link SchemaRegistryAvroSerializerBuilder} instance.
+     * @return updated {@link SchemaRegistryApacheAvroSerializerBuilder} instance.
      */
-    public SchemaRegistryAvroSerializerBuilder schemaRegistryAsyncClient(
+    public SchemaRegistryApacheAvroSerializerBuilder schemaRegistryAsyncClient(
         SchemaRegistryAsyncClient schemaRegistryAsyncClient) {
         this.schemaRegistryAsyncClient = schemaRegistryAsyncClient;
         return this;
     }
 
     /**
-     * Instantiates SchemaRegistry Avro serializer.
+     * Creates a new instance of Schema Registry serializer.
      *
-     * @return {@link SchemaRegistryAvroSerializer} instance
+     * @return A new instance of {@link SchemaRegistryApacheAvroSerializer}.
      *
      * @throws NullPointerException if {@link #schemaRegistryAsyncClient(SchemaRegistryAsyncClient)} is {@code null}
      *     or {@link #schemaGroup(String) schemaGroup} is {@code null}.
      * @throws IllegalArgumentException if credential is not set.
      */
-    public SchemaRegistryAvroSerializer buildSerializer() {
+    public SchemaRegistryApacheAvroSerializer buildSerializer() {
         final boolean isAutoRegister = autoRegisterSchemas != null && autoRegisterSchemas;
         final boolean useAvroSpecificReader = avroSpecificReader == null
             ? AVRO_SPECIFIC_READER_DEFAULT : avroSpecificReader;
         final Schema.Parser parser = new Schema.Parser();
         final AvroSerializer codec = new AvroSerializer(useAvroSpecificReader, parser,
             EncoderFactory.get(), DecoderFactory.get());
+        final SerializerOptions options = new SerializerOptions(schemaGroup, isAutoRegister, MAX_CACHE_SIZE);
 
-        return new SchemaRegistryAvroSerializer(schemaRegistryAsyncClient, codec, schemaGroup, isAutoRegister);
+        return new SchemaRegistryApacheAvroSerializer(schemaRegistryAsyncClient, codec, options);
     }
 }

sdk/schemaregistry/azure-data-schemaregistry-avro/src/main/java/com/azure/data/schemaregistry/avro/SerializerOptions.java

@@ -0,0 +1,56 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.data.schemaregistry.avro;
+
+import com.azure.core.annotation.Immutable;
+
+/**
+ * Package-private class that holds additional options when creating serializer.
+ */
+@Immutable
+class SerializerOptions {
+    private final boolean autoRegisterSchemas;
+    private final int maxCacheSize;
+    private final String schemaGroup;
+
+    /**
+     * Creates a new instance.
+     *
+     * @param schemaGroup Optional schema group when registering a schema is required.
+     * @param autoRegisterSchemas {@code true} to register schema if it does not exist, {@code false} otherwise.
+     * @param maxCacheSize The maximum cache size for the serializer.
+     */
+    SerializerOptions(String schemaGroup, boolean autoRegisterSchemas, int maxCacheSize) {
+        this.schemaGroup = schemaGroup;
+        this.autoRegisterSchemas = autoRegisterSchemas;
+        this.maxCacheSize = maxCacheSize;
+    }
+
+    /**
+     * Gets whether or not to auto-register schemas.
+     *
+     * @return {@code true} to register schema if it does not exist; {@code false} otherwise.
+     */
+    boolean autoRegisterSchemas() {
+        return autoRegisterSchemas;
+    }
+
+    /**
+     * Gets the maximum cache size.
+     *
+     * @return The maximum cache size.
+     */
+    int getMaxCacheSize() {
+        return maxCacheSize;
+    }
+
+    /**
+     * Gets the schema group to register schemas against.
+     *
+     * @return The schema group.
+     */
+    String getSchemaGroup() {
+        return schemaGroup;
+    }
+}

sdk/schemaregistry/azure-data-schemaregistry-avro/src/samples/java/com/azure/data/schemaregistry/avro/ReadmeSamples.java

@@ -25,18 +25,18 @@
 public class ReadmeSamples {
 
     /**
-     * Sample to demonstrate creation of {@link SchemaRegistryAvroSerializer}.
-     * @return The {@link SchemaRegistryAvroSerializer}.
+     * Sample to demonstrate creation of {@link SchemaRegistryApacheAvroSerializer}.
+     * @return The {@link SchemaRegistryApacheAvroSerializer}.
      */
-    public SchemaRegistryAvroSerializer createAvroSchemaRegistrySerializer() {
+    public SchemaRegistryApacheAvroSerializer createAvroSchemaRegistrySerializer() {
         TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();
 
         SchemaRegistryAsyncClient schemaRegistryAsyncClient = new SchemaRegistryClientBuilder()
             .fullyQualifiedNamespace("{schema-registry-endpoint")
             .credential(tokenCredential)
             .buildAsyncClient();
 
-        SchemaRegistryAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryAvroSerializerBuilder()
+        SchemaRegistryApacheAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryApacheAvroSerializerBuilder()
             .schemaRegistryAsyncClient(schemaRegistryAsyncClient)
             .schemaGroup("{schema-group}")
             .buildSerializer();
@@ -48,7 +48,7 @@ public SchemaRegistryAvroSerializer createAvroSchemaRegistrySerializer() {
      * Serialize a strongly-typed object into avro payload compatible with schema registry.
      */
     public void serializeSample() {
-        SchemaRegistryAvroSerializer schemaRegistryAvroSerializer = createAvroSchemaRegistrySerializer();
+        SchemaRegistryApacheAvroSerializer schemaRegistryAvroSerializer = createAvroSchemaRegistrySerializer();
 
         PlayingCard playingCard = new PlayingCard();
         playingCard.setPlayingCardSuit(PlayingCardSuit.SPADES);
@@ -65,7 +65,7 @@ public void serializeSample() {
      * Deserialize avro payload compatible with schema registry into a strongly-type object.
      */
     public void deserializeSample() {
-        SchemaRegistryAvroSerializer schemaRegistryAvroSerializer = createAvroSchemaRegistrySerializer();
+        SchemaRegistryApacheAvroSerializer schemaRegistryAvroSerializer = createAvroSchemaRegistrySerializer();
         InputStream inputStream = getSchemaRegistryAvroData();
         PlayingCard playingCard = schemaRegistryAvroSerializer.deserialize(inputStream,
             TypeReference.createInstance(PlayingCard.class));

sdk/schemaregistry/azure-data-schemaregistry-avro/src/samples/java/com/azure/data/schemaregistry/avro/SchemaRegistryAvroDeserilizationSample.java

@@ -36,7 +36,7 @@ public static void main(String[] args) throws IOException {
 
         // Create the serializer instance by configuring the serializer with the schema registry client and
         // enabling auto registering of new schemas
-        SchemaRegistryAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryAvroSerializerBuilder()
+        SchemaRegistryApacheAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryApacheAvroSerializerBuilder()
             .schemaRegistryAsyncClient(schemaRegistryAsyncClient)
             .schemaGroup("{schema-group}")
             .avroSpecificReader(true)

sdk/schemaregistry/azure-data-schemaregistry-avro/src/samples/java/com/azure/data/schemaregistry/avro/SchemaRegistryAvroSerializationSample.java

@@ -14,7 +14,7 @@
 import java.io.OutputStream;
 
 /**
- * Sample to demonstrate using {@link SchemaRegistryAvroSerializer} for serialization of data.
+ * Sample to demonstrate using {@link SchemaRegistryApacheAvroSerializer} for serialization of data.
  */
 public class SchemaRegistryAvroSerializationSample {
     /**
@@ -34,7 +34,7 @@ public static void main(String[] args) {
 
         // Create the serializer instance by configuring the serializer with the schema registry client and
         // enabling auto registering of new schemas
-        SchemaRegistryAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryAvroSerializerBuilder()
+        SchemaRegistryApacheAvroSerializer schemaRegistryAvroSerializer = new SchemaRegistryApacheAvroSerializerBuilder()
             .schemaRegistryAsyncClient(schemaRegistryAsyncClient)
             .schemaGroup("{schema-group}")
             .avroSpecificReader(true)