Wrap Avro exceptions and other fixes (Azure#27581)

* Wrap Avro exceptions

* Remove preamble support

* Add SchemaId to exception

* Add troubleshooting docs

* Update

* Update

* Improve error handling for BinaryContent types without public parameterless constructor

* Recording files

* Fix validation

* Improve error handling

* Update snippet

* Typo

* Fix

Author: JoshLove-msft

sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md

@@ -131,7 +131,30 @@ Employee deserializedEmployee = await serializer.DeserializeAsync<Employee>(cont
 
 ## Troubleshooting
 
-Information on troubleshooting steps will be provided as potential issues are discovered.
+If you encounter errors when communicating with the Schema Registry service, these errors will be thrown as a [RequestFailedException][request_failed_exception]. The serializer will only communicate with the service the first time it encounters a schema (when serializing) or a schema ID (when deserializing). Any errors related to serialization to Avro, or deserialization from Avro, will be thrown as a `AvroSerializationException`. The `InnerException` property will contain the underlying exception that was thrown from the Apache Avro library. When deserializing, the `SerializedSchemaId` property will contain the schema ID corresponding to the serialized data. Using our `Employee` schema example, if we add an `Employee_V2` model that adds a new required field, this would not be compatible with `Employee`. If the data we are attempting to deserialize may contain a schema that would not be compatible with our `Employee_V2` model, then we might write code like the following:
+
+```C# Snippet:SchemaRegistryAvroException
+try
+{
+    Employee_V2 employeeV2 = await serializer.DeserializeAsync<Employee_V2>(content);
+}
+catch (AvroSerializationException exception)
+{
+    // When this exception occurs when deserializing, the exception message will contain the schema ID that was used to
+    // serialize the data.
+    Console.WriteLine(exception);
+
+    // We might also want to look up the specific schema from Schema Registry so that we can log the schema definition
+    if (exception.SerializedSchemaId != null)
+    {
+        SchemaRegistrySchema schema = await client.GetSchemaAsync(exception.SerializedSchemaId);
+        Console.WriteLine(schema.Definition);
+    }
+}
+```
+
+In general, any invalid Avro schemas would probably be caught during testing, but such schemas will also result in a `AvroSerializationException` being thrown when attempting to serialize using an invalid writer schema, or deserialize when using an invalid reader schema.
+
 
 ## Next steps
 
@@ -169,3 +192,4 @@ This project has adopted the [Microsoft Open Source Code of Conduct][code_of_con
 [specific_record]: https://avro.apache.org/docs/current/api/csharp/html/interfaceAvro_1_1Specific_1_1ISpecificRecord.html
 [azure_sub]: https://azure.microsoft.com/free/dotnet/
 [azure_schema_registry]: https://aka.ms/schemaregistry
+[request_failed_exception]: https://docs.microsoft.com/dotnet/api/azure.requestfailedexception?view=azure-dotnet

sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs

@@ -1,5 +1,15 @@
 namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro
 {
+    public partial class AvroSerializationException : System.Exception
+    {
+        public AvroSerializationException() { }
+        protected AvroSerializationException(System.Runtime.Serialization.SerializationInfo info, System.Runtime.Serialization.StreamingContext context) { }
+        public AvroSerializationException(string message) { }
+        public AvroSerializationException(string message, System.Exception innerException) { }
+        public AvroSerializationException(string message, string serializedSchemaId, System.Exception innerException) { }
+        public string SerializedSchemaId { get { throw null; } set { } }
+        public override void GetObjectData(System.Runtime.Serialization.SerializationInfo info, System.Runtime.Serialization.StreamingContext context) { }
+    }
     public partial class SchemaRegistryAvroSerializer
     {
         public SchemaRegistryAvroSerializer(Azure.Data.SchemaRegistry.SchemaRegistryClient client, string groupName, Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.SchemaRegistryAvroSerializerOptions options = null) { }

sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/AvroSerializationException.cs

@@ -0,0 +1,71 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Runtime.Serialization;
+using Azure.Core;
+
+namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro
+{
+    /// <summary>
+    /// Represents an exception that is thrown when Avro serialization or deserialization fails.
+    /// </summary>
+    [Serializable]
+    public class AvroSerializationException : Exception
+    {
+        /// <summary>
+        /// The Schema Registry schema Id related to the serialized data that caused the <see cref="AvroSerializationException"/>.
+        /// </summary>
+        public string SerializedSchemaId { get; set; }
+
+        /// <summary>
+        /// Initializes a new instance of <see cref="AvroSerializationException"/>.
+        /// </summary>
+        public AvroSerializationException() : this(null, null)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of <see cref="AvroSerializationException"/>.
+        /// </summary>
+        /// <param name="message">The error message that explains the reason for the exception.</param>
+        public AvroSerializationException(string message) : this(message, null)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of <see cref="AvroSerializationException"/>.
+        /// </summary>
+        /// <param name="message">The error message that explains the reason for the exception.</param>
+        /// <param name="innerException">The exception that is the cause of the current exception, or a null reference if no inner exception is specified.</param>
+        public AvroSerializationException(string message, Exception innerException) : this(message, null, innerException)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of <see cref="AvroSerializationException"/>.
+        /// </summary>
+        /// <param name="message">The error message that explains the reason for the exception.</param>
+        /// <param name="serializedSchemaId">The Schema Registry schema Id related to the <see cref="AvroSerializationException"/>.</param>
+        /// <param name="innerException">The exception that is the cause of the current exception, or a null reference if no inner exception is specified.</param>
+        public AvroSerializationException(string message, string serializedSchemaId, Exception innerException) : base(message, innerException)
+        {
+            SerializedSchemaId = serializedSchemaId;
+        }
+
+        /// <inheritdoc />
+        protected AvroSerializationException(SerializationInfo info, StreamingContext context) : base(info, context)
+        {
+            SerializedSchemaId = info.GetString(nameof(SerializedSchemaId));
+        }
+
+        /// <inheritdoc />
+        public override void GetObjectData(SerializationInfo info, StreamingContext context)
+        {
+            Argument.AssertNotNull(info, nameof(info));
+
+            info.AddValue(nameof(SerializedSchemaId), SerializedSchemaId);
+            base.GetObjectData(info, context);
+        }
+    }
+}

sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroSerializer.cs

@@ -9,8 +9,6 @@
 using Azure.Data.SchemaRegistry;
 using System;
 using System.IO;
-using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure;
@@ -29,7 +27,6 @@ public class SchemaRegistryAvroSerializer
         private readonly SchemaRegistryAvroSerializerOptions _options;
         private const string AvroMimeType = "avro/binary";
         private const int CacheCapacity = 128;
-        private static readonly Encoding Utf8Encoding = new UTF8Encoding(false);
 
         /// <summary>
         /// Initializes new instance of <see cref="SchemaRegistryAvroSerializer"/>.
@@ -41,10 +38,6 @@ public SchemaRegistryAvroSerializer(SchemaRegistryClient client, string groupNam
             _options = options;
         }
 
-        private static readonly byte[] EmptyRecordFormatIndicator = { 0, 0, 0, 0 };
-        private const int RecordFormatIndicatorLength = 4;
-        private const int SchemaIdLength = 32;
-        private const int PayloadStartPosition = RecordFormatIndicatorLength + SchemaIdLength;
         private readonly LruCache<string, Schema> _idToSchemaMap = new(CacheCapacity);
         private readonly LruCache<Schema, string> _schemaToIdMap = new(CacheCapacity);
 
@@ -125,12 +118,18 @@ internal async ValueTask<BinaryContent> SerializeInternalAsync(
             bool async,
             CancellationToken cancellationToken)
         {
+            messageType ??= typeof(BinaryContent);
+            if (messageType.GetConstructor(Type.EmptyTypes) == null)
+            {
+                throw new InvalidOperationException(
+                    $"The type {messageType} must have a public parameterless constructor in order to use it as the 'MessageContent' type to serialize to.");
+            }
+            var message = (BinaryContent)Activator.CreateInstance(messageType);
+
             (string schemaId, BinaryData bd) = async
                 ? await SerializeInternalAsync(data, dataType, true, cancellationToken).ConfigureAwait(false)
                 : SerializeInternalAsync(data, dataType, false, cancellationToken).EnsureCompleted();
 
-            messageType ??= typeof(BinaryContent);
-            var message = (BinaryContent)Activator.CreateInstance(messageType);
             message.Data = bd;
             message.ContentType = $"{AvroMimeType}+{schemaId}";
             return message;
@@ -146,23 +145,31 @@ internal async ValueTask<BinaryContent> SerializeInternalAsync(
             dataType ??= value?.GetType() ?? typeof(object);
 
             var supportedType = GetSupportedTypeOrThrow(dataType);
-            var writer = GetWriterAndSchema(value, supportedType, out var schema);
 
-            using Stream stream = new MemoryStream();
-            var binaryEncoder = new BinaryEncoder(stream);
+            try
+            {
+                var writer = GetWriterAndSchema(value, supportedType, out var schema);
 
-            writer.Write(value, binaryEncoder);
-            binaryEncoder.Flush();
-            stream.Position = 0;
-            BinaryData data = BinaryData.FromStream(stream);
+                using Stream stream = new MemoryStream();
+                var binaryEncoder = new BinaryEncoder(stream);
 
-            if (async)
-            {
-                return (await GetSchemaIdAsync(schema, true, cancellationToken).ConfigureAwait(false), data);
+                writer.Write(value, binaryEncoder);
+                binaryEncoder.Flush();
+                stream.Position = 0;
+                BinaryData data = BinaryData.FromStream(stream);
+
+                if (async)
+                {
+                    return (await GetSchemaIdAsync(schema, true, cancellationToken).ConfigureAwait(false), data);
+                }
+                else
+                {
+                    return (GetSchemaIdAsync(schema, false, cancellationToken).EnsureCompleted(), data);
+                }
             }
-            else
+            catch (AvroException ex)
             {
-                return (GetSchemaIdAsync(schema, false, cancellationToken).EnsureCompleted(), data);
+                throw new AvroSerializationException("An error occurred while attempting to serialize to Avro.", ex);
             }
         }
 
@@ -300,35 +307,18 @@ private async ValueTask<object> DeserializeMessageDataInternalAsync(
             Argument.AssertNotNull(data, nameof(data));
             Argument.AssertNotNull(contentType, nameof(contentType));
 
-            string schemaId;
-            // Back Compat for first preview
-            ReadOnlyMemory<byte> memory = data.ToMemory();
-            byte[] recordFormatIdentifier = null;
-            if (memory.Length >= RecordFormatIndicatorLength)
+            string[] contentTypeArray = contentType.ToString().Split('+');
+            if (contentTypeArray.Length != 2)
             {
-                recordFormatIdentifier = memory.Slice(0, RecordFormatIndicatorLength).ToArray();
+                throw new FormatException("Content type was not in the expected format of MIME type + schema ID");
             }
-            if (recordFormatIdentifier != null && recordFormatIdentifier.SequenceEqual(EmptyRecordFormatIndicator))
+
+            if (contentTypeArray[0] != AvroMimeType)
             {
-                byte[] schemaIdBytes = memory.Slice(RecordFormatIndicatorLength, SchemaIdLength).ToArray();
-                schemaId = Utf8Encoding.GetString(schemaIdBytes);
-                data = new BinaryData(memory.Slice(PayloadStartPosition, memory.Length - PayloadStartPosition));
+                throw new InvalidOperationException("An avro serializer may only be used on content that is of 'avro/binary' type");
             }
-            else
-            {
-                string[] contentTypeArray = contentType.ToString().Split('+');
-                if (contentTypeArray.Length != 2)
-                {
-                    throw new FormatException("Content type was not in the expected format of MIME type + schema ID");
-                }
 
-                if (contentTypeArray[0] != AvroMimeType)
-                {
-                    throw new InvalidOperationException("An avro serializer may only be used on content that is of 'avro/binary' type");
-                }
-
-                schemaId = contentTypeArray[1];
-            }
+            string schemaId = contentTypeArray[1];
 
             if (async)
             {
@@ -351,27 +341,63 @@ private async ValueTask<object> DeserializeInternalAsync(
             SupportedType supportedType = GetSupportedTypeOrThrow(dataType);
 
             Schema writerSchema;
-            if (async)
+            try
             {
-                writerSchema = await GetSchemaByIdAsync(schemaId, true, cancellationToken).ConfigureAwait(false);
+                if (async)
+                {
+                    writerSchema = await GetSchemaByIdAsync(schemaId, true, cancellationToken).ConfigureAwait(false);
+                }
+                else
+                {
+                    writerSchema = GetSchemaByIdAsync(schemaId, false, cancellationToken).EnsureCompleted();
+                }
             }
-            else
+            catch (SchemaParseException ex)
             {
-                writerSchema = GetSchemaByIdAsync(schemaId, false, cancellationToken).EnsureCompleted();
+                throw new AvroSerializationException(
+                    $"An error occurred while attempting to parse the schema (schema ID: {schemaId}) that was used to serialize the Avro. " +
+                    $"Make sure that the schema represents valid Avro.",
+                    schemaId,
+                    ex);
             }
 
-            var binaryDecoder = new BinaryDecoder(data.ToStream());
+            Schema readerSchema;
+            object returnInstance = null;
+            try
+            {
+                if (supportedType == SupportedType.SpecificRecord)
+                {
+                    returnInstance = Activator.CreateInstance(dataType);
+                    readerSchema = ((ISpecificRecord)returnInstance).Schema;
+                }
+                else
+                {
+                    readerSchema = writerSchema;
+                }
+            }
+            catch (SchemaParseException ex)
+            {
+                throw new AvroSerializationException(
+                    "An error occurred while attempting to parse the schema that you are attempting to deserialize the data with. " +
+                    "Make sure that the schema represents valid Avro.",
+                    schemaId,
+                    ex);
+            }
 
-            if (supportedType == SupportedType.SpecificRecord)
+            try
             {
-                object returnInstance = Activator.CreateInstance(dataType);
-                DatumReader<object> reader = GetReader(writerSchema, ((ISpecificRecord)returnInstance).Schema, SupportedType.SpecificRecord);
+                var binaryDecoder = new BinaryDecoder(data.ToStream());
+                DatumReader<object> reader = GetReader(writerSchema, readerSchema, supportedType);
                 return reader.Read(reuse: returnInstance, binaryDecoder);
             }
-            else
+            catch (AvroException ex)
             {
-                DatumReader<object> reader = GetReader(writerSchema, writerSchema, supportedType);
-                return reader.Read(reuse: null, binaryDecoder);
+                throw new AvroSerializationException(
+                    "An error occurred while attempting to deserialize " +
+                    $"Avro that was serialized with schemaId: {schemaId}. The schema used to deserialize the data may not be compatible with the schema that was used" +
+                    $"to serialize the data. Please ensure that the schemas are compatible.",
+                    schemaId,
+                    ex);
             }
         }