npm package: @securepipeco/kafka-schema-registry-client
Description: Kafka Schema Registy client.
Author: anton.v.rublev@gmail.com
Latest version: 0.3.1

# @securepipe.co/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 @securepipe.co/kafka-schema-registry-client
```
or
```
yarn add @securepipe.co/kafka-schema-registry-client
```

## Usage

### Create registry client

```js
const { SchemaRegistry } = require('@securepipe.co/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.

```js
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.

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

#### Fetch and set active latest versions of all registered topics

```js
await schemaRegistry.useSchemas();
```

#### Fetch and set active latest versions for specific topics

```js
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.

```js
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.

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

### Deserializing messsage

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

```js
 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