# datasource-kafka

An Apollo [`DataSource`](https://github.com/apollographql/apollo-server/tree/f63fc5adcd092db682c0d62b904e9077d206ebd8/packages/apollo-datasource)
for interacting with [Kafka](https://kafka.apache.org/)

## Usage

### Configuring

```js
const config = {
    serviceName: 'graphql-raed',
    kafkaBrokers: ['broker1', 'broker2', 'broker3'],
};

const kafka = new KafkaDataSource(config);
```

You can also pass an existing `kafkajs` instance to the constructor:

```js
import { Kafka } from 'kafkajs';

const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['kafka1:9092', 'kafka2:9092'],
});

const kafka = new KafkaDataSource({ kafka });
```

### Extending `KafkaDataSource`

```js
import { KafkaDataSource } from '@theorchard/datasource-kafka';

class Kafka extends KafkaDataSource {
    async sendMyMessage({ message = 'Hello Kafka world' }) {
        return this.queueMessage({
            topic: 'cdc.topic.helloworld',
            key: 'optional, typically some uuid or compound index',
            payload: {
                message,
            },
        });
    }
}

export default Kafka;
```

As long as your `config` object has the Kafka properties as defined above, this should work without adding a `constructor` to the class.

### How to use Schema registry?

```js
const config = {
    serviceName: 'lambda-test',
    kafkaBrokers: ['broker1', 'broker2', 'broker3'],
    schemaRegistryUrl: 'https://dev-schema-registry.dev.theorchard.io',
};

const kafkaDataSource = new KafkaDataSource(config);
result = await kafkaDataSource.queueMessage({
    topic,
    payload,
    withSchemaRegistry: true,
});
```

Note:

-   When you set withSchemaRegistry=true This assumes that you already have a schema registered with name `<topicname>-value`. It will throw an Error if that is not there.
-   If your message does not honor the schema it will throw ConfluentSchemaRegistryValidationError