schema-registry-ajv

An Ajv instance wrapper for @kafkajs/confluent-schema-registry.

Description

The library is a wrapper for the Ajv instance, which is in turn used in conjunction with @kafkajs/confluent-schema-registry.

Key Features

• Resolves the Ajv "schema with id ... already exists" error that occurs at high RPS during communication with Kafka when using @kafkajs/confluent-schema-registry.
• Allows retrieval of Ajv validation errors for the payload data by pulling them up to the application level.
• Resolves the TypeError when passing an Ajv instance to the SchemaRegistry constructor options.

Bonus Features

• Saves you from having to handle the schema ID.

Installation

Install the library:

npm install @sylvanas-cry/schema-registry-ajv

Install peer dependencies:

npm install \
  @kafkajs/confluent-schema-registry \
  ajv \
  ajv-formats \
  axios

Usage

Constructing and initialization:

const builder = new SchemaRegistryAjvBuilder({
  // Ajv class you want to use. Check "$schema" attribute of your schemas
  ajvClass: Ajv2020,
  // Optional Ajv native format options
  ajvFormats: ['date-time'],
  // Optional Ajv custom format options
  ajvCustomFormats: [{ name: 'custom-format', format: (value) => !!value }],
  // Schema Registry options
  schemaRegistry: {
    // Schema Registry URI
    host: 'https://schema-registry.example.com:8081',
    // Schema subject
    subject: 'foo.bar.baz.event.baz-done.v0-value',
    // Optional version of schema you want to use (can be number or 'latest')
    version: 'latest',
  },
});

const { ajvInstance, getSchemaId, getErrors } = await builder.build();

const schemaRegistry = new SchemaRegistry({ 
  host: 'https://schema-registry.example.com:8081',
}, {
  [SchemaType.JSON]: {
    ajvInstance, // <-- put your new safe Ajv instance here
  },
});

Somewhere in your code:

// Publishing to Kafka topic
try {
  // To encode Kafka message value
  const value = await schemaRegistry.encode(getSchemaId(), event);

  // ...
} catch (err) {
  // To get Ajv validation errors
  const validationErorrs = getErrors();

  // ...
}

Changes and Release Notes

CHANGELOG.md

License

The scripts and documentation in this project are released under the MIT License.