@namaste/kafka-schema-registry-client

    0.3.0 • Public • Published

    @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

    1. 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

    1. 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

    1. message - message to be serialized
    2. 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

    1. 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

    1. 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

    1. serializedMessage - message to be deserialized
    2. topic - topic the message was published to

    Install

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

    DownloadsWeekly Downloads

    1

    Version

    0.3.0

    License

    MIT

    Unpacked Size

    22.5 kB

    Total Files

    9

    Last publish

    Collaborators

    • antonrublev
    • jpdentone
    • mysza