[Schema Registry Avro] Support backward compatibility for old preamble format (Azure#20091)

* [Schema Registry Avro] Support backward compatibility for old preamble format

* fix linting

* address feedback

Author: deyaaeldeen

sdk/schemaregistry/schema-registry-avro/CHANGELOG.md

@@ -4,7 +4,7 @@
 
 ### Features Added
 
-- The serializer APIs have been revamped to work on messages instead of buffers where the payload is the pure encoded-data. The schema ID became part of the content type of that message. This change will improve the experience of using this encoder with the other messaging clients (e.g. Event Hubs, Service Bus, and Event Grid clients).
+- The encoder APIs have been revamped to work on messages instead of buffers where the payload is the pure encoded-data. The schema ID became part of the content type of that message. This change will improve the experience of using this encoder with the other messaging clients (e.g. Event Hubs, Service Bus, and Event Grid clients). The encoder also supports decoding messages with payloads that follow the old format where the schema ID was part of the payload.
 - `decodeMessageData` now supports decoding using a different but compatible schema
 
 ### Breaking Changes

sdk/schemaregistry/schema-registry-avro/README.md

@@ -58,6 +58,16 @@ by setting the `messageAdapter` option in the constructor with a corresponding
 message producer and consumer. Azure messaging client libraries export default
 adapters for their message types.
 
+### Backward Compatibility
+
+The encoder v1.0.0-beta.5 and under encodes data into binary arrays. Starting from
+v1.0.0-beta.6, the encoder returns messages instead that contain the encoded payload.
+For a smooth transition to using the newer versions, the encoder also supports
+decoding messages with payloads that follow the old format where the schema ID
+is part of the payload.
+
+This backward compatibility is temporary and will be removed in v1.0.0.
+
 ## Examples
 
 ### Encode and decode an `@azure/event-hubs`'s `EventData`

sdk/schemaregistry/schema-registry-avro/src/schemaRegistryAvroEncoder.ts

@@ -95,7 +95,7 @@ export class SchemaRegistryAvroEncoder<MessageT = MessageWithMetadata> {
     options: DecodeMessageDataOptions = {}
   ): Promise<unknown> {
     const { schema: readerSchema } = options;
-    const { body, contentType } = getPayloadAndContent(message, this.messageAdapter);
+    const { body, contentType } = convertMessage(message, this.messageAdapter);
     const buffer = Buffer.from(body);
     const writerSchemaId = getSchemaId(contentType);
     const writerSchema = await this.getSchema(writerSchemaId);
@@ -201,18 +201,63 @@ function getSchemaId(contentType: string): string {
   return contentTypeParts[1];
 }
 
-function getPayloadAndContent<MessageT>(
+/**
+ * Tries to decode a body in the preamble format. If that does not succeed, it
+ * returns it as is.
+ * @param body - The message body
+ * @param contentType - The message content type
+ * @returns a message with metadata
+ */
+function convertPayload(body: Uint8Array, contentType: string): MessageWithMetadata {
+  try {
+    return tryReadingPreambleFormat(Buffer.from(body));
+  } catch (_e: unknown) {
+    return {
+      body,
+      contentType,
+    };
+  }
+}
+
+function convertMessage<MessageT>(
   message: MessageT,
-  messageAdapter?: MessageAdapter<MessageT>
+  adapter?: MessageAdapter<MessageT>
 ): MessageWithMetadata {
-  const messageConsumer = messageAdapter?.consumeMessage;
+  const messageConsumer = adapter?.consumeMessage;
   if (messageConsumer) {
-    return messageConsumer(message);
+    const { body, contentType } = messageConsumer(message);
+    return convertPayload(body, contentType);
   } else if (isMessageWithMetadata(message)) {
-    return message;
+    return convertPayload(message.body, message.contentType);
   } else {
     throw new Error(
-      `Either the messageConsumer option should be defined or the message should have body and contentType fields`
+      `Expected either a message adapter to be provided to the encoder or the input message to have body and contentType fields`
     );
   }
 }
+
+/**
+ * Maintains backward compatability by supporting the encoded value format created
+ * by earlier beta serializers
+ * @param buffer - The input buffer
+ * @returns a message that contains the body and content type with the schema ID
+ */
+function tryReadingPreambleFormat(buffer: Buffer): MessageWithMetadata {
+  const FORMAT_INDICATOR = 0;
+  const SCHEMA_ID_OFFSET = 4;
+  const PAYLOAD_OFFSET = 36;
+  if (buffer.length < PAYLOAD_OFFSET) {
+    throw new RangeError("Buffer is too small to have the correct format.");
+  }
+  const format = buffer.readUInt32BE(0);
+  if (format !== FORMAT_INDICATOR) {
+    throw new TypeError(`Buffer has unknown format indicator.`);
+  }
+  const schemaIdBuffer = buffer.slice(SCHEMA_ID_OFFSET, PAYLOAD_OFFSET);
+  const schemaId = schemaIdBuffer.toString("utf-8");
+  const payloadBuffer = buffer.slice(PAYLOAD_OFFSET);
+  return {
+    body: payloadBuffer,
+    contentType: `${avroMimeType}+${schemaId}`,
+  };
+}

sdk/schemaregistry/schema-registry-avro/test/schemaRegistryAvroEncoder.spec.ts

@@ -189,4 +189,22 @@ describe("SchemaRegistryAvroEncoder", function () {
       /no matching field for default-less com.azure.schemaregistry.samples.AvroUser.age/
     );
   });
+
+  it("decodes from the old format", async () => {
+    const registry = createTestRegistry();
+    const schemaId = await registerTestSchema(registry);
+    const encoder = await createTestEncoder(false, registry);
+    const payload = testAvroType.toBuffer(testValue);
+    const buffer = Buffer.alloc(36 + payload.length);
+
+    buffer.write(schemaId, 4, 32, "utf-8");
+    payload.copy(buffer, 36);
+    assert.deepStrictEqual(
+      await encoder.decodeMessageData({
+        body: buffer,
+        contentType: "avro/binary+000",
+      }),
+      testValue
+    );
+  });
 });