Adds common exception for serialization related failures (Azure#28093)

* Add SchemaRegistryAvroException.

* Update serialization to throw AvroSerializationException.

* Update CHANGELOG.

* Updating exceptions to be thrown during serialization. Add documentation.

* Updating documentation for exceptions.

* Adding code snippets.

* Adding code snippets.

* Rename exception.

* Rename helper method to deserialize.

* Update sdk/schemaregistry/azure-data-schemaregistry-apacheavro/CHANGELOG.md

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

Author: conniey

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/CHANGELOG.md

@@ -7,8 +7,10 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+
 - Fixed a bug that caused deserialize operation to throw `SchemaParseException` when multiple messages with same schema 
   were deserialized (https://github.com/Azure/azure-sdk-for-java/issues/27602).
+- Wrap Apache Avro exceptions with new exception type, `SchemaRegistryApacheAvroException`.
 
 ### Other Changes
 

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/README.md

@@ -28,17 +28,16 @@ and deserialization.
 ```
 [//]: # ({x-version-update-end})
 
-### Create `SchemaRegistryAvroSerializer` instance
+### Create `SchemaRegistryApacheAvroSerializer` instance
 
-The `SchemaRegistryAvroSerializer` instance is the main class that provides APIs for serializing and
+The `SchemaRegistryApacheAvroSerializer` instance is the main class that provides APIs for serializing and
 deserializing avro data format. The avro schema is stored and retrieved from the Schema Registry service
 through the `SchemaRegistryAsyncClient`. So, before we create the serializer, we should create the client.
 
 #### Create `SchemaRegistryAsyncClient` with Azure Active Directory Credential
 
 In order to interact with the Azure Schema Registry service, you'll need to create an instance of the
-`SchemaRegistryAsyncClient` class through the `SchemaRegistryClientBuilder`. You will need an **endpoint** and an
-**API key** to instantiate a client object.
+`SchemaRegistryAsyncClient` class through the `SchemaRegistryClientBuilder`. You will need the Schema Registry **endpoint**.
 
 You can authenticate with Azure Active Directory using the [Azure Identity library][azure_identity]. Note that regional
 endpoints do not support AAD authentication. Create a [custom subdomain][custom_subdomain] for your resource in order to
@@ -81,22 +80,13 @@ SchemaRegistryApacheAvroSerializer serializer = new SchemaRegistryApacheAvroSeri
 
 ## Key concepts
 
-### ObjectSerializer
-This library provides a serializer, `SchemaRegistryAvroSerializer`, that implements the `ObjectSerializer` interface.
-This allows a developer to use this serializer in any Java Azure SDKs that utilize `ObjectSerializer`. The
+This library provides a serializer, `SchemaRegistryApacheAvroSerializer`. The
 `SchemaRegistryAvroSerializer` utilizes a `SchemaRegistryAsyncClient` to construct messages using a wire format
 containing schema information such as a schema ID.
 
 This serializer requires the Apache Avro library. The payload types accepted by this serializer include
 [GenericRecord][generic_record] and [SpecificRecord][specific_record].
 
-### Wire Format
-The serializer in this library creates messages in a wire format. The format is the following:
-
-- Bytes [0-3] – record format indicator – currently is \x00\x00\x00\x00
-- Bytes [4-35] – UTF-8 GUID, identifying the schema in a Schema Registry instance
-- Bytes [36-end] – serialized payload bytes
-
 ## Examples
 
 * [Serialize](#serialize)

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/src/main/java/com/azure/data/schemaregistry/apacheavro/AvroSerializer.java

@@ -23,7 +23,6 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
@@ -134,14 +133,15 @@ Schema parseSchemaString(String schemaString) {
      * Returns A byte[] containing Avro encoding of object parameter.
      *
      * @param object Object to be encoded into byte stream
+     * @param schemaId Identifier of the schema trying to be encoded.
      *
      * @return A set of bytes that represent the object.
      *
      * @throws IllegalArgumentException If the object is not a serializable type.
      * @throws IllegalStateException if the object could not be serialized to an object stream or there was a
      *     runtime exception during serialization.
      */
-    <T> byte[] encode(T object) {
+    <T> byte[] serialize(T object, String schemaId) {
         final Schema schema = getSchema(object);
 
         try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
@@ -162,7 +162,8 @@ <T> byte[] encode(T object) {
             return outputStream.toByteArray();
         } catch (IOException | RuntimeException e) {
             // Avro serialization can throw AvroRuntimeException, NullPointerException, ClassCastException, etc
-            throw logger.logExceptionAsError(new IllegalStateException("Error serializing Avro message", e));
+            throw logger.logExceptionAsError(new SchemaRegistryApacheAvroException(
+                "An error occurred while attempting to serialize to Avro.", e, schemaId));
         }
     }
 
@@ -172,7 +173,7 @@ <T> byte[] encode(T object) {
      *
      * @return deserialized object
      */
-    <T> T decode(ByteBuffer contents, byte[] schemaBytes, TypeReference<T> typeReference) {
+    <T> T deserialize(ByteBuffer contents, byte[] schemaBytes, TypeReference<T> typeReference) {
         Objects.requireNonNull(contents, "'bytes' must not be null.");
         Objects.requireNonNull(schemaBytes, "'schemaBytes' must not be null.");
 
@@ -185,7 +186,7 @@ <T> T decode(ByteBuffer contents, byte[] schemaBytes, TypeReference<T> typeRefer
             try {
                 return messageDecoder.decode(contents);
             } catch (IOException e) {
-                throw logger.logExceptionAsError(new UncheckedIOException(
+                throw logger.logExceptionAsError(new SchemaRegistryApacheAvroException(
                     "Unable to deserialize Avro schema object using binary message decoder.", e));
             }
         } else {
@@ -196,7 +197,8 @@ <T> T decode(ByteBuffer contents, byte[] schemaBytes, TypeReference<T> typeRefer
                     return reader.read(null, decoderFactory.binaryDecoder(input, null));
                 }
             } catch (IOException | RuntimeException e) {
-                throw logger.logExceptionAsError(new IllegalStateException("Error deserializing raw Avro message.", e));
+                throw logger.logExceptionAsError(new SchemaRegistryApacheAvroException(
+                    "Error deserializing raw Avro message.", e));
             }
         }
     }

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/src/main/java/com/azure/data/schemaregistry/apacheavro/SchemaRegistryApacheAvroException.java

@@ -0,0 +1,73 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.data.schemaregistry.apacheavro;
+
+import com.azure.core.exception.AzureException;
+
+/**
+ * Represents an exception that is thrown when Avro serialization or deserialization fails.
+ */
+public class SchemaRegistryApacheAvroException extends AzureException {
+    private final String schemaId;
+
+    /**
+     * Initializes a new instance.
+     *
+     * @param message The exception message.
+     */
+    public SchemaRegistryApacheAvroException(String message) {
+        super(message);
+
+        this.schemaId = null;
+    }
+
+    /**
+     * Initializes a new instance.
+     *
+     * @param message The exception message.
+     * @param cause The {@link Throwable} which caused the creation of this exception.
+     */
+    public SchemaRegistryApacheAvroException(String message, Throwable cause) {
+        super(message, cause);
+
+        this.schemaId = null;
+    }
+
+    /**
+     * Initializes a new instance.
+     *
+     * @param message The exception message.
+     * @param cause The {@link Throwable} which caused the creation of this exception.
+     * @param enableSuppression Whether suppression is enabled or disabled.
+     * @param writableStackTrace Whether the exception stack trace will be filled in.
+     */
+    public SchemaRegistryApacheAvroException(String message, Throwable cause, boolean enableSuppression,
+        boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+
+        this.schemaId = null;
+    }
+
+    /**
+     * Initializes a new instance.
+     *
+     * @param message The exception message.
+     * @param cause The {@link Throwable} which caused the creation of this exception.
+     * @param schemaId The id of the schema being processed when this exception occurred. {@code null} if there was
+     *     none.
+     */
+    public SchemaRegistryApacheAvroException(String message, Throwable cause, String schemaId) {
+        super(message, cause);
+        this.schemaId = schemaId;
+    }
+
+    /**
+     * Gets the schema id that was associated with this exception.
+     *
+     * @return The schema id associated with teh exception. {@code null} if there was no schema id.
+     */
+    public String getSchemaId() {
+        return schemaId;
+    }
+}