@namaste/kafka-schema-registry-client

@namaste/kafka-schema-registry-client

Kafka Avro schema registry client that supports:

• Serializing messages
• Publishing locally defined schemas to registry
• Selecting schema versions to use for serializing messages
• Deserializing messages
• Pre-fetching latest schemas for topics from registry

Installation

npm i --save @namaste/kafka-schema-registry-client

or

yarn add @namaste/kafka-schema-registry-client

Usage

Create registry client

const { SchemaRegistry } = require('@namaste/kafka-schema-registry-client');

const connectionConfig = {
  url: 'https://my.fake.schema.registry.url:3030',
  auth: {
    username: 'fakeusername',
    password: 'fakepassword',
  },
};

const schemaRegistry = new SchemaRegistry(connectionConfig);

Publish locally defined avro schemas

Note: Schemas are only published if changed. If schema version has already been published, it is skipped. Note: Published schemas are also cached and set to be used for serialization.

const topicSchemaDefinitions = {
  'topic': {
    key: {
      name: 'testKey',
      type: 'string',
      namespace: 'registry.test',
    },
    value: {
      name: 'testValue',
      type: 'record',
      namespace: 'registry.test',
      fields: [
        {
          name: 'text',
          type: 'string',
        },
        {
          name: 'newField',
          type: ['null', 'string'],
          default: null,
        },
      ],
    },
  }
};
await schemaRegistry.publishSchemas(topicSchemaDefinitions);

Caching schemas from registry and setting them to be used for serialization

Note: Usefull for producers when schemas are not defined locally or should not be automatically published by producer.

Specifying versions for topics

Specific schema versions will be fetched and set active.

await schemaRegistry.useSchemas({
  'topic_name': {
    key: 'latest',
    value: '2',
  }
});

Fetch and set active latest versions of all registered topics

await schemaRegistry.useSchemas();

Fetch and set active latest versions for specific topics

await schemaRegistry.useSchemas(['topic_name_1', 'topic_name_2']);

Serializing message

Note: Schemas should be set to be used for serialization by either of two methods described above prior to serialization. Note: Schema version used for serialization is recorded in message headers.

const message = {
  key: 'key text',
  value: {
    text: 'some text',
    number: 4,
    array: ['test1', 'test2'],
    newField: 'more text',
  },
  partition: 2,
  headers: {},
  timestamp: '2019-01-01T22:22:22.000Z',
};
const serializedMessage = schemaRegistry.serializeMessage(message, 'topic');

Pre-fetching schemas for deserialization

Note: This is optional as schemas that has not been fetched yet are fetched on deserialization attempt.

const topicsToPrefetchSchemasFor = ['topicname'];
await schemaRegistry.fetchLatestSchemas(topicsToPrefetchSchemasFor);

Deserializing messsage

Note: Schema version used for deserialization is defined in message headers.

 const deserializedMessage = await schemaRegistry.deserializeMessage(serializedMessage, 'topic');

API

constructor

Creates new instance of schema registry.

Parameters

• connectionConfig - object that has following shape:

• url - schema registry URL
• auth - basic auth config, must have:
• auth.username
• auth.password

publishSchemas(topicSchemaDefinitions): Promise<void>

Publishes local shemas to registry and sets them to be used for messages serialization.

Paramerters

• topicSchemaDefinitions - object that has following shape:

• [topic_name]:
• [topic_name].key - optional. Avro schema definition for message key.
• [topic_name].value - optional. Avro schema definition for message value.

serializeMessage(message, 'topic'): serializedMessage

Serializes message using key and value schemas for the topic. Schema versions to be used are defined with either publishSchemas or useSchemas methods.

Parameters

• message - message to be serialized
• topic - topic the message is going to be published to

async useSchemas([topicVersionDefinitions]): Promise<void>

Fetches schema versions for specified topics from registry and sets them to be used for messages serialization.

Paramerters

• topicVersionDefinitions - Optional. Either a string array of topic names or an object that specifies key and value versions:

When object, must have following shape:

• [topic_name]:
• [topic_name].key - optional. Schema version string. Can be version number or string latest.
• [topic_name].value - optional. Schema version string. Can be version number or string latest.

When array, must list topic names.

Note: If no topicVersionDefinitions are passed to useSchemas, schema registry will fetch and set as active latest verions for all subjects (keys and values for all topics).

async fetchLatestSchemas(topicsToPrefetchSchemasFor): Promise<void>

Fetches latest schemas for specified topics from registry.

Paramerters

• topicsToPrefetchSchemasFor - array of topic name strings

async deserializeMessage(serializedMessage, 'topic'): Promise<deserializedMessage>

Deserializes message using key and value schemas for the topic. Schema versions to be used are retreived from message headers.

Parameters

• serializedMessage - message to be deserialized
• topic - topic the message was published to