From 43c57c9cd5aa7882ec37d455d838f73ede03b2b2 Mon Sep 17 00:00:00 2001 From: svozza Date: Sun, 23 Aug 2026 17:19:55 +0000 Subject: [PATCH] fix(kafka): lazily import Avro and Protobuf codecs so JSON-only consumers bundle The consumer used string-literal dynamic imports for the Avro and Protobuf deserializers. Bundlers such as esbuild statically resolve literal dynamic imports and follow their transitive `avro-js`/`protobufjs` imports, so a primitive/JSON-only handler failed to bundle unless both packages were installed, and always shipped both codecs. Load each codec through a non-literal specifier inside an async factory (`createDeserializer`) so bundlers leave it as a runtime import that resolves from `node_modules` only when the format is used. `deserialize` stays synchronous (it runs inside record getters) and no top-level `await` is introduced, keeping the CommonJS build valid. Also declare `avro-js` and `protobufjs` as optional peer dependencies, matching the existing parser peer-dependency pattern. Closes #5557 --- docs/features/kafka.md | 4 + packages/kafka/package.json | 8 + packages/kafka/src/consumer.ts | 8 +- packages/kafka/src/deserializer/avro.ts | 49 +++-- packages/kafka/src/deserializer/protobuf.ts | 180 ++++++++++-------- packages/kafka/src/types/types.ts | 10 + .../tests/unit/deserializer.avro.test.ts | 10 +- .../tests/unit/deserializer.protobuf.test.ts | 10 +- 8 files changed, 182 insertions(+), 97 deletions(-) diff --git a/docs/features/kafka.md b/docs/features/kafka.md index 3d96b7f1d3..2c93b8c073 100644 --- a/docs/features/kafka.md +++ b/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 diff --git a/packages/kafka/package.json b/packages/kafka/package.json index 353a0f812e..31786b8ae0 100644 --- a/packages/kafka/package.json +++ b/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": [ diff --git a/packages/kafka/src/consumer.ts b/packages/kafka/src/consumer.ts index 6034221ae2..aeef65be6d 100644 --- a/packages/kafka/src/consumer.ts +++ b/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.` diff --git a/packages/kafka/src/deserializer/avro.ts b/packages/kafka/src/deserializer/avro.ts index 19d2888035..4145c2d0b7 100644 --- a/packages/kafka/src/deserializer/avro.ts +++ b/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 => { + 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 }; diff --git a/packages/kafka/src/deserializer/protobuf.ts b/packages/kafka/src/deserializer/protobuf.ts index 14b0f58120..c941f2dc33 100644 --- a/packages/kafka/src/deserializer/protobuf.ts +++ b/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 = ( - data: string, - messageType: ProtobufMessage, - schemaMetadata: SchemaMetadata -): T => { - const buffer = Buffer.from(data, 'base64'); - try { - if (schemaMetadata.schemaId === undefined) { - return messageType.decode(buffer, buffer.length); - } +const createDeserializer = async (): Promise => { + 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 = ( + data: string, + messageType: ProtobufMessage, + 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 }; diff --git a/packages/kafka/src/types/types.ts b/packages/kafka/src/types/types.ts index 4b53a6e06c..3a88f72af6 100644 --- a/packages/kafka/src/types/types.ts +++ b/packages/kafka/src/types/types.ts @@ -251,6 +251,14 @@ type Deserializer = ( schemaMetadata?: SchemaMetadata ) => unknown; +type AvroDeserializer = (data: string, schema: string) => unknown; + +type ProtobufDeserializer = ( + data: string, + messageType: ProtobufMessage, + 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, diff --git a/packages/kafka/tests/unit/deserializer.avro.test.ts b/packages/kafka/tests/unit/deserializer.avro.test.ts index 7edccd5960..1f2210efce 100644 --- a/packages/kafka/tests/unit/deserializer.avro.test.ts +++ b/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>; + + beforeAll(async () => { + deserialize = await createDeserializer(); + }); + it('returns avro deserialised value', async () => { // Prepare const message = '0g8MTGFwdG9wUrgehes/j0A='; diff --git a/packages/kafka/tests/unit/deserializer.protobuf.test.ts b/packages/kafka/tests/unit/deserializer.protobuf.test.ts index 07d9fcd4e1..f9046ccdd1 100644 --- a/packages/kafka/tests/unit/deserializer.protobuf.test.ts +++ b/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>; + + beforeAll(async () => { + deserialize = await createDeserializer(); + }); + it('throws when protobuf serialise fails', () => { // Prepare const data = 'COkHEgZMYXB0b3AZUrgehes/j0A=';