Kafka batch processing (#333)

* make buckets parameter required for histogram metric

* Added batch processing support to Kafka consumer and removed messageType routing in handlers

* mention Kafka in README

* refactoring and improvements

* tests fix

* tests fix

* removed redundant event type from test permission consumers

* condition fix

* types improvements

* tests fix

* linting

* consumer closing improvement

* refactoring

* added DeserializedMessage type

* reduced coverage for kafka test branches

* updated kafka version

* removed default batch size and timeout values

* renamed folder

* renamed test file

* collect batches per topic and partition

* added README for kafka

* @platformatic/kafka version update

Author: kjamrog

README.md

@@ -16,6 +16,7 @@ It consists of the following submodules:
 * `@message-queue-toolkit/amqp` - AMQP 0-9-1 (Advanced Message Queuing Protocol), used e. g. by RabbitMQ
 * `@message-queue-toolkit/sqs` - SQS (AWS Simple Queue Service)
 * `@message-queue-toolkit/sns` - SNS (AWS Simple Notification Service)
+* `@message-queue-toolkit/kafka` - Kafka (in development)
 
 ## Basic Usage
 
@@ -29,7 +30,7 @@ They implement the following public methods:
     * `options`, composed by
         * `messageSchemas` – the `zod` schemas for all supported messages;
         * `messageTimestampField` - which field in the message contains the message creation date (by default it is `timestamp`). This field needs to be a `Date` object or ISO-8601 date string, if your message doesn't contain it the library will add one automatically to avoid infinite loops on consumer;
-        * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for resolving the correct schema for validation
+        * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for resolving the correct schema for validation. **Note:** It is not supported for Kafka publisher
         * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`.
         * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`;
         * `policyConfig` - SQS only - configuration for queue access policies (see [SQS Policy Configuration](#sqs-policy-configuration) for more information);
@@ -93,7 +94,7 @@ Multi-schema consumers support multiple message types via handler configs. They
     * `dependencies` – a set of dependencies depending on the protocol;
     * `options`, composed by
         * `handlers` – configuration for handling each of the supported message types. See "Multi-schema handler definition" for more details;
-        * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for routing the message to the correct handler;
+        * `messageTypeField` - which field in the message describes the type of a message. This field needs to be defined as `z.literal` in the schema and is used for routing the message to the correct handler; **Note:** It is not supported for Kafka consumer
         * `messageTimestampField` - which field in the message contains the message creation date (by default it is `timestamp`). This field needs to be a `Date` object or an ISO-8601 date string;
         * `maxRetryDuration` - how long (in seconds) the message should be retried due to the `retryLater` result before marking it as consumed (and sending to DLQ, if one is configured). This is used to avoid infinite loops. Default is 4 days;
         * `queueName`; (for SNS publishers this is a misnomer which actually refers to a topic name)

packages/kafka/README.md

@@ -1,7 +1,40 @@
-# @message-queue-toolkit/kafka
+# Kafka
 
-In development
+This library provides utilities for implementing Kafka consumers and publishers.
+While following the same patterns as other message broker implementations,
+Kafka's unique characteristics require some specific adaptations in the publisher and consumer definitions.
 
+> **_NOTE:_** Check [README.md](../../README.md) for transport-agnostic library documentation.
 
-## Requirements
-- Node.js >= 22.14.0
+## Publishers
+
+Use `AbstractKafkaPublisher` as a base class for publisher implementation.
+
+See [test publisher](test/publisher/PermissionPublisher.ts) for an example of implementation.
+
+## Consumers
+
+Use `AbstractKafkaConsumer` as a base class for consumer implementation.
+
+See [test consumer](test/consumer/PermissionConsumer.ts) for an example of implementation.
+
+## Batch Processing
+
+Kafka supports batch processing for improved throughput. To enable it, set `batchProcessingEnabled` to `true` and configure `batchProcessingOptions`.
+
+When batch processing is enabled, message handlers receive an array of messages instead of a single message.
+
+### Configuration Options
+
+- `batchSize` - Maximum number of messages per batch
+- `timeoutMilliseconds` - Maximum time to wait for a batch to fill before processing
+
+### How It Works
+
+Messages are buffered per topic-partition combination. Batches are processed when either:
+- The buffer reaches the configured `batchSize`
+- The `timeoutMilliseconds` timeout is reached since the first message was added
+
+After successful batch processing, the offset of the last message in the batch is committed.
+
+See [test batch consumer](test/consumer/PermissionBatchConsumer.ts) for an example of implementation.