fix(kafka): lazily import Avro and Protobuf codecs so JSON-only consumers bundle

docs/features/kafka.md

@@ -66,6 +66,10 @@ Depending on the schema types you want to use, install the library and the corre
 	npm install @aws-lambda-powertools/kafka protobufjs
 	```
 
+`avro-js` and `protobufjs` are declared as optional peer dependencies (`avro-js@^1.12.1`, `protobufjs@^8.7.2`).
+They are only imported when you configure an Avro or Protobuf schema, so a JSON-only consumer bundles without them installed.
+If you already have one of these installed at an incompatible version you may see a peer-dependency warning; align it with the supported range to clear it.
+
 Additionally, if you want to use output parsing with [Standard Schema](https://github.com/standard-schema/standard-schema), you can install [any of the supported libraries](https://standardschema.dev/#what-schema-libraries-implement-the-spec), for example: Zod, Valibot, or ArkType.
 
 ### Required resources

packages/kafka/package.json

@@ -53,6 +53,8 @@
   },
   "peerDependencies": {
     "arktype": "^2.2.3",
+    "avro-js": "^1.12.1",
+    "protobufjs": "^8.7.2",
     "valibot": "^1.4.2",
     "zod": "^3.25.0 || ^4.0.0"
   },
@@ -65,6 +67,12 @@
     },
     "arktype": {
       "optional": true
+    },
+    "avro-js": {
+      "optional": true
+    },
+    "protobufjs": {
+      "optional": true
     }
   },
   "files": [

packages/kafka/src/consumer.ts

@@ -120,12 +120,12 @@ const getDeserializer = async (type?: string) => {
     return deserializeJson as Deserializer;
   }
   if (type === 'protobuf') {
-    const deserializer = await import('./deserializer/protobuf.js');
-    return deserializer.deserialize as Deserializer;
+    const { createDeserializer } = await import('./deserializer/protobuf.js');
+    return (await createDeserializer()) as Deserializer;
   }
   if (type === 'avro') {
-    const deserializer = await import('./deserializer/avro.js');
-    return deserializer.deserialize as Deserializer;
+    const { createDeserializer } = await import('./deserializer/avro.js');
+    return (await createDeserializer()) as Deserializer;
   }
   throw new KafkaConsumerDeserializationError(
     `Unsupported deserialization type: ${type}. Supported types are: json, avro, protobuf.`

packages/kafka/src/deserializer/avro.ts

@@ -1,22 +1,43 @@
-import avro from 'avro-js';
 import { KafkaConsumerDeserializationError } from '../errors.js';
+import type { AvroDeserializer } from '../types/types.js';
+
+let deserializer: AvroDeserializer | undefined;
 
 /**
- * Deserialize an Avro message from a base64-encoded string using the provided Avro schema.
+ * Create an Avro deserializer, importing `avro-js` on first use.
  *
- * @param data - The base64-encoded string representing the Avro binary data.
- * @param schema - The Avro schema as a JSON string.
+ * The `avro-js` dependency is only required when deserializing Avro messages,
+ * so it's resolved at runtime through a non-literal specifier. This keeps
+ * bundlers (e.g. esbuild) from eagerly including `avro-js` - and failing when
+ * it isn't installed - in builds that never use Avro.
  */
-const deserialize = (data: string, schema: string) => {
-  try {
-    const type = avro.parse(schema);
-    const buffer = Buffer.from(data, 'base64');
-    return type.fromBuffer(buffer);
-  } catch (error) {
-    throw new KafkaConsumerDeserializationError(
-      `Failed to deserialize Avro message: ${error}, message: ${data}, schema: ${schema}`
-    );
+const createDeserializer = async (): Promise<AvroDeserializer> => {
+  if (deserializer !== undefined) {
+    return deserializer;
   }
+
+  const moduleName = 'avro-js';
+  const { default: avro }: typeof import('avro-js') = await import(moduleName);
+
+  /**
+   * Deserialize an Avro message from a base64-encoded string using the provided Avro schema.
+   *
+   * @param data - The base64-encoded string representing the Avro binary data.
+   * @param schema - The Avro schema as a JSON string.
+   */
+  deserializer = (data: string, schema: string) => {
+    try {
+      const type = avro.parse(schema);
+      const buffer = Buffer.from(data, 'base64');
+      return type.fromBuffer(buffer);
+    } catch (error) {
+      throw new KafkaConsumerDeserializationError(
+        `Failed to deserialize Avro message: ${error}, message: ${data}, schema: ${schema}`
+      );
+    }
+  };
+
+  return deserializer;
 };
 
-export { deserialize };
+export { createDeserializer };

packages/kafka/src/deserializer/protobuf.ts

@@ -1,94 +1,124 @@
-import { BufferReader, type Message } from 'protobufjs';
 import { KafkaConsumerDeserializationError } from '../errors.js';
-import type { ProtobufMessage, SchemaMetadata } from '../types/types.js';
+import type {
+  ProtobufDeserializer,
+  ProtobufMessage,
+  SchemaMetadata,
+} from '../types/types.js';
 
-/**
- * Default order of varint types used in Protobuf to attempt deserializing Confluent Schema Registry messages.
- */
-const varintOrder: Array<'int32' | 'sint32'> = ['int32', 'sint32'];
+let deserializer: ProtobufDeserializer | undefined;
 
 /**
- * Deserialize a Protobuf message from a base64-encoded string.
+ * Create a Protobuf deserializer, importing `protobufjs` on first use.
  *
- * @template T - The type of the deserialized message object.
- *
- * @param data - The base64-encoded string representing the Protobuf binary data.
- * @param messageType - The Protobuf message type definition - see {@link Message | `Message`} from {@link https://www.npmjs.com/package/protobufjs | `protobufjs`}.
+ * The `protobufjs` dependency is only required when deserializing Protobuf
+ * messages, so it's resolved at runtime through a non-literal specifier. This
+ * keeps bundlers (e.g. esbuild) from eagerly including `protobufjs` - and
+ * failing when it isn't installed - in builds that never use Protobuf. See
+ * {@link https://www.npmjs.com/package/protobufjs | `protobufjs`}.
  */
-const deserialize = <T>(
-  data: string,
-  messageType: ProtobufMessage<T>,
-  schemaMetadata: SchemaMetadata
-): T => {
-  const buffer = Buffer.from(data, 'base64');
-  try {
-    if (schemaMetadata.schemaId === undefined) {
-      return messageType.decode(buffer, buffer.length);
-    }
+const createDeserializer = async (): Promise<ProtobufDeserializer> => {
+  if (deserializer !== undefined) {
+    return deserializer;
+  }
+
+  const moduleName = 'protobufjs';
+  const { BufferReader }: typeof import('protobufjs') = await import(
+    moduleName
+  );
+
+  /**
+   * Default order of varint types used in Protobuf to attempt deserializing Confluent Schema Registry messages.
+   */
+  const varintOrder: Array<'int32' | 'sint32'> = ['int32', 'sint32'];
+
+  /**
+   * Clip the Confluent Schema Registry buffer to remove the index bytes.
+   *
+   * @param buffer - The buffer to clip.
+   * @param intType - The type of the integer to read from the buffer, either 'int32' or 'sint32'.
+   */
+  const clipConfluentSchemaRegistryBuffer = (
+    buffer: Buffer,
+    intType: 'int32' | 'sint32'
+  ) => {
+    const reader = new BufferReader(buffer);
     /**
-     * If `schemaId` is longer than 10 chars, it's an UUID, otherwise it's a numeric ID.
-     *
-     * When this is the case, we know the schema is coming from Glue Schema Registry,
-     * and the first byte of the buffer is a magic byte that we need to remove before
-     * decoding the message.
+     * Read the first varint byte to get the index count or 0.
+     * Doing so, also advances the reader position to the next byte after the index count.
      */
-    if (schemaMetadata.schemaId.length > 10) {
-      // remove the first byte from the buffer
-      const reader = new BufferReader(buffer);
-      reader.uint32();
-      return messageType.decode(reader);
-    }
-  } catch (error) {
-    throw new KafkaConsumerDeserializationError(
-      `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${JSON.stringify(messageType)}`
-    );
-  }
+    const indexCount = intType === 'int32' ? reader.int32() : reader.sint32();
+    // Skip the index bytes
+    reader.skip(indexCount);
+    return reader;
+  };
 
   /**
-   * If schemaId is numeric, inferred from its length, we know it's coming from Confluent Schema Registry,
-   * so we need to remove the MessageIndex bytes.
-   * We don't know the type of the index, so we try both `int32` and `sint32`. If both fail, we throw an error.
+   * Deserialize a Protobuf message from a base64-encoded string.
+   *
+   * @template T - The type of the deserialized message object.
+   *
+   * @param data - The base64-encoded string representing the Protobuf binary data.
+   * @param messageType - The Protobuf message type definition - see the `Message` type from {@link https://www.npmjs.com/package/protobufjs | `protobufjs`}.
    */
-  try {
-    const newBuffer = clipConfluentSchemaRegistryBuffer(buffer, varintOrder[0]);
-    return messageType.decode(newBuffer);
-  } catch (error) {
+  deserializer = <T>(
+    data: string,
+    messageType: ProtobufMessage<T>,
+    schemaMetadata: SchemaMetadata
+  ): T => {
+    const buffer = Buffer.from(data, 'base64');
     try {
-      const newBuffer = clipConfluentSchemaRegistryBuffer(
-        buffer,
-        varintOrder[1]
-      );
-      const decoded = messageType.decode(newBuffer);
-      // swap varint order if the first attempt failed so we can use the correct one for subsequent messages
-      varintOrder.reverse();
-      return decoded;
-    } catch {
+      if (schemaMetadata.schemaId === undefined) {
+        return messageType.decode(buffer, buffer.length);
+      }
+      /**
+       * If `schemaId` is longer than 10 chars, it's an UUID, otherwise it's a numeric ID.
+       *
+       * When this is the case, we know the schema is coming from Glue Schema Registry,
+       * and the first byte of the buffer is a magic byte that we need to remove before
+       * decoding the message.
+       */
+      if (schemaMetadata.schemaId.length > 10) {
+        // remove the first byte from the buffer
+        const reader = new BufferReader(buffer);
+        reader.uint32();
+        return messageType.decode(reader);
+      }
+    } catch (error) {
       throw new KafkaConsumerDeserializationError(
         `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${JSON.stringify(messageType)}`
       );
     }
-  }
-};
 
-/**
- * Clip the Confluent Schema Registry buffer to remove the index bytes.
- *
- * @param buffer - The buffer to clip.
- * @param intType - The type of the integer to read from the buffer, either 'int32' or 'sint32'.
- */
-const clipConfluentSchemaRegistryBuffer = (
-  buffer: Buffer,
-  intType: 'int32' | 'sint32'
-) => {
-  const reader = new BufferReader(buffer);
-  /**
-   * Read the first varint byte to get the index count or 0.
-   * Doing so, also advances the reader position to the next byte after the index count.
-   */
-  const indexCount = intType === 'int32' ? reader.int32() : reader.sint32();
-  // Skip the index bytes
-  reader.skip(indexCount);
-  return reader;
+    /**
+     * If schemaId is numeric, inferred from its length, we know it's coming from Confluent Schema Registry,
+     * so we need to remove the MessageIndex bytes.
+     * We don't know the type of the index, so we try both `int32` and `sint32`. If both fail, we throw an error.
+     */
+    try {
+      const newBuffer = clipConfluentSchemaRegistryBuffer(
+        buffer,
+        varintOrder[0]
+      );
+      return messageType.decode(newBuffer);
+    } catch (error) {
+      try {
+        const newBuffer = clipConfluentSchemaRegistryBuffer(
+          buffer,
+          varintOrder[1]
+        );
+        const decoded = messageType.decode(newBuffer);
+        // swap varint order if the first attempt failed so we can use the correct one for subsequent messages
+        varintOrder.reverse();
+        return decoded;
+      } catch {
+        throw new KafkaConsumerDeserializationError(
+          `Failed to deserialize Protobuf message: ${error}, message: ${data}, messageType: ${JSON.stringify(messageType)}`
+        );
+      }
+    }
+  };
+
+  return deserializer;
 };
 
-export { deserialize };
+export { createDeserializer };

packages/kafka/src/types/types.ts

@@ -251,6 +251,14 @@ type Deserializer = (
   schemaMetadata?: SchemaMetadata
 ) => unknown;
 
+type AvroDeserializer = (data: string, schema: string) => unknown;
+
+type ProtobufDeserializer = <T>(
+  data: string,
+  messageType: ProtobufMessage<T>,
+  schemaMetadata: SchemaMetadata
+) => T;
+
 type DeserializeOptions = {
   value: string | null;
   deserializer: Deserializer;
@@ -259,11 +267,13 @@ type DeserializeOptions = {
 };
 
 export type {
+  AvroDeserializer,
   ConsumerRecord,
   ConsumerRecords,
   DeserializeOptions,
   Deserializer,
   MSKEvent,
+  ProtobufDeserializer,
   ProtobufMessage,
   Record,
   RecordHeader,

packages/kafka/tests/unit/deserializer.avro.test.ts

@@ -1,8 +1,14 @@
-import { describe, expect, it } from 'vitest';
-import { deserialize } from '../../src/deserializer/avro.js';
+import { beforeAll, describe, expect, it } from 'vitest';
+import { createDeserializer } from '../../src/deserializer/avro.js';
 import { KafkaConsumerDeserializationError } from '../../src/errors.js';
 
 describe('Avro Deserializer: ', () => {
+  let deserialize: Awaited<ReturnType<typeof createDeserializer>>;
+
+  beforeAll(async () => {
+    deserialize = await createDeserializer();
+  });
+
   it('returns avro deserialised value', async () => {
     // Prepare
     const message = '0g8MTGFwdG9wUrgehes/j0A=';

packages/kafka/tests/unit/deserializer.protobuf.test.ts

@@ -1,11 +1,17 @@
 import type { Message } from 'protobufjs';
-import { describe, expect, it } from 'vitest';
-import { deserialize } from '../../src/deserializer/protobuf.js';
+import { beforeAll, describe, expect, it } from 'vitest';
+import { createDeserializer } from '../../src/deserializer/protobuf.js';
 import { KafkaConsumerDeserializationError } from '../../src/errors.js';
 import type { ProtobufMessage } from '../../src/types/types.js';
 import { Product } from '../protos/product.generated.js';
 
 describe('Protobuf deserialiser: ', () => {
+  let deserialize: Awaited<ReturnType<typeof createDeserializer>>;
+
+  beforeAll(async () => {
+    deserialize = await createDeserializer();
+  });
+
   it('throws when protobuf serialise fails', () => {
     // Prepare
     const data = 'COkHEgZMYXB0b3AZUrgehes/j0A=';