| Method
| Type |
Description
|
getMessages |
KafkaMessage[] |
Get the raw messages from Kafka in the batch |
getConsumer |
KafkaConsumer |
Get the consumer instance |
getResolveOffset |
function |
Get the resolve offset method |
getHeartbeat |
function |
Get the heartbeat method |
getPause |
function |
Get the pause method |
getCommitOffsetsIfNecessary |
function |
Get the commit offsets if necessary method |
getUncommittedOffsets |
OffsetsByTopicPartition |
Get the uncommitted offsets |
getIsRunning |
boolean |
Indicate if the consumer is still running |
getIsStale |
boolean |
Indicate if the consumer is stale |
### Client
The `KafkaBatchClient` is exactly the same as the `KafkaClient` from the `@nestjs/microservices` package, except that `client.send` method is removed from the client as batch messages should not be used for `request-response` communication. On top of that, `KafkaBatchClient` also have the capability to [publish batch messages](https://kafka.js.org/docs/producing#producing-messages) or [publish to multiple topics](https://kafka.js.org/docs/producing#producing-to-multiple-topics) just like in `kafkajs`.
```typescript
@Module({
imports: [
ClientsModule.register([{
name: 'KAFKA_BATCH_CLIENT',
customClass: KafkaBatchClient,
options: {
client: {
brokers: ['localhost:52800', 'localhost:52801'],
},
consumer: {
groupId: 'test',
heartbeatInterval: 5000,
sessionTimeout: 30000,
},
},
}]),
],
})
export class AppModule {}
```
Then you can inject and use the `KafkaBatchClient` in your service as follow
```typescript
@Injectable()
export class AppService {
constructor(
@Inject('KAFKA_BATCH_CLIENT')
private kafkaClient: KafkaBatchClient,
) {}
async eventToBatch() {
this.kafkaClient.emit('test', { example: 'data'});
}
async publishBatch() {
// equivalent to kafkajs producer.send
this.kafkaClient.emitBatch('test', [{
example: 'data1'
}, {
example: 'data2'
}])
}
async publishBatchTopics() {
// will publish to two topics, topic1 and topic2
// equivalent to kafkajs producer.publishBatch
this.kafkaClient.emitBatchTopics([{
pattern: 'topic1',
data: [{ example: 'data11' }, { example: 'data12' }]
}, {
pattern: 'topic2',
data: [{ example: 'data21' }, { example: 'data22' }]
}])
}
}
```
Calling `send` with the `KafkaBatchClient` will result in an error.
```typescript
this.kafkaClient.send('send', { data: 'data'}); // Error
```
