Add first test for promisified API: consumer.

Adds test helpers and jest.

Author: milindl

MIGRATION.md

@@ -139,7 +139,6 @@
 
 #### Semantic and Per-Method Changes
 
-* `sendBatch` is not supported (YET). However, the actual batching semantics are handled by librdkafka.
 * Changes to `send`:
   * `acks`, `compression` and `timeout` are not set on a per-send basis. Rather, they must be configured in the configuration.
     Before:
@@ -178,6 +177,7 @@
     ```
 
   * Error-handling for a failed `send` is stricter. While sending multiple messages, even if one of the messages fails, the method throws an error.
+* `sendBatch` is supported. However, the actual batching semantics are handled by librdkafka, and it just acts as a wrapper around `send` (See `send` for changes).
 
 ### Consumer
 
@@ -219,7 +219,10 @@
 #### Semantic and Per-Method Changes
 
 
- * While passing a list of topics to `subscribe`, the `fromBeginning` property is not supported. Instead, the property `auto.offset.reset` needs to be used.
+ * Changes to subscribe:
+  * Regex flags are ignored while passing a topic subscription (like 'i' or 'g').
+  * Subscribe must be called after `connect`.
+  * While passing a list of topics to `subscribe`, the `fromBeginning` property is not supported. Instead, the property `auto.offset.reset` needs to be used.
    Before:
     ```javascript
       const kafka = new Kafka({ /* ... */ });
@@ -229,7 +232,6 @@
       await consumer.connect();
       await consumer.subscribe({ topics: ["topic"], fromBeginning: true});
     ```
-
    After:
     ```javascript
       const kafka = new Kafka({ /* ... */ });

Makefile

@@ -13,7 +13,8 @@ endif
 NODE ?= node
 CPPLINT ?= cpplint.py
 BUILDTYPE ?= Release
-TESTS = "test/**/*.js"
+TESTS = $(ls test/producer/*.js test/*.js test/tools/*.js)
+PROMISIFIED_TESTS = "test/promisified"
 E2E_TESTS = $(wildcard e2e/*.spec.js)
 TEST_REPORTER =
 TEST_OUTPUT =
@@ -24,7 +25,7 @@ CONFIG_OUTPUTS = \
 
 CPPLINT_FILES = $(wildcard src/*.cc src/*.h)
 CPPLINT_FILTER = -legal/copyright
-JSLINT_FILES = lib/*.js test/*.js e2e/*.js
+JSLINT_FILES = lib/*.js test/*.js e2e/*.js lib/kafkajs/*.js
 
 PACKAGE = $(shell node -pe 'require("./package.json").name.split("/")[1]')
 VERSION = $(shell node -pe 'require("./package.json").version')
@@ -58,6 +59,7 @@ $(CONFIG_OUTPUTS): node_modules/.dirstamp binding.gyp
 
 test: node_modules/.dirstamp
 	@./node_modules/.bin/mocha --ui exports $(TEST_REPORTER) $(TESTS) $(TEST_OUTPUT)
+	@./node_modules/.bin/jest --ci --runInBand $(PROMISIFIED_TESTS)
 
 check: node_modules/.dirstamp
 	@$(NODE) util/test-compile.js

lib/kafkajs/_admin.js

@@ -90,6 +90,12 @@ class Admin {
    * @returns {Promise<void>} Resolves when disconnect is complete, rejects on error.
    */
   async disconnect() {
+    /* Not yet connected - no error. */
+    if (this.#state == AdminState.INIT) {
+      return;
+    }
+
+    /* Already disconnecting, or disconnected. */
     if (this.#state >= AdminState.DISCONNECTING) {
       return;
     }

lib/kafkajs/_consumer.js

@@ -342,16 +342,48 @@ class Consumer {
 
   /**
    * Subscribes the consumer to the given topics.
-   * @param {import("../../types/kafkajs").ConsumerSubscribeTopics} subscription
+   * @param {import("../../types/kafkajs").ConsumerSubscribeTopics | import("../../types/kafkajs").ConsumerSubscribeTopic} subscription
    */
   async subscribe(subscription) {
-    if (typeof subscription.fromBeginning == 'boolean') {
+    if (this.#state !== ConsumerState.CONNECTED) {
+      throw new error.KafkaJSError('Subscribe can only be called while connected.', { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    if (typeof subscription.fromBeginning === 'boolean') {
       throw new error.KafkaJSError(
-        'fromBeginning is not supported by subscribe(), but must be passed as rdKafka properties to the consumer',
-        { code: error.ErrorCodes.ERR__INVALID_ARG });
+        'fromBeginning is not supported by subscribe(), but must be passed as an rdKafka property to the consumer.',
+        { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
     }
 
-    this.#internalClient.subscribe(subscription.topics);
+    if (!Object.hasOwn(subscription, 'topics') && !Object.hasOwn(subscription, 'topic')) {
+      throw new error.KafkaJSError('Either topics or topic must be specified.', { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    let topics = [];
+    if (subscription.topic) {
+      topics.push(subscription.topic);
+    } else if (Array.isArray(subscription.topics)) {
+      topics = subscription.topics;
+    } else {
+      throw new error.KafkaJSError('topics must be an object of the type ConsumerSubscribeTopics.', { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    topics = topics.map(topic => {
+      if (typeof topic === 'string') {
+        return topic;
+      } else if (topic instanceof RegExp) {
+        // Flags are not supported, and librdkafka only considers a regex match if the first character of the regex is ^.
+        const regexSource = topic.source;
+        if (regexSource.charAt(0) !== '^')
+          return '^' + regexSource;
+        else
+          return regexSource;
+      } else {
+        throw new error.KafkaJSError('Invalid topic ' + topic + ' (' + typeof topic + '), the topic name has to be a String or a RegExp', { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+    });
+
+    this.#internalClient.subscribe(topics);
   }
 
   async stop() {
@@ -541,10 +573,18 @@ class Consumer {
    * @returns {Promise<void>} a promise that resolves when the consumer has disconnected.
    */
   async disconnect() {
-    if (this.#state === ConsumerState.INIT) {
-      throw new error.KafkaJSError('Disconnect can only be called once consumer is connected.', { code: error.ErrorCodes.ERR__STATE });
+    /* Not yet connected - no error. */
+    if (this.#state == ConsumerState.INIT) {
+      return;
     }
 
+    /* TODO: We should handle a case where we are connecting, we should
+     * await the connection and then schedule a disconnect. */
+
+    /* Already disconnecting, or disconnected. */
+    if (this.#state >= ConsumerState.DISCONNECTING) {
+      return;
+    }
     if (this.#state >= ConsumerState.DISCONNECTING) {
       return;
     }

lib/kafkajs/_producer.js

@@ -240,6 +240,15 @@ class Producer {
    * @returns {Promise<void>} Resolves when disconnect is complete, rejects on error.
    */
   async disconnect() {
+    /* Not yet connected - no error. */
+    if (this.#state == ProducerState.INIT) {
+      return;
+    }
+
+    /* TODO: We should handle a case where we are connecting, we should
+     * await the connection and then schedule a disconnect. */
+
+    /* Already disconnecting, or disconnected. */
     if (this.#state >= ProducerState.DISCONNECTING) {
       return;
     }
@@ -477,6 +486,44 @@ class Producer {
     }
     return ret;
   }
+
+  /**
+    * sendBatch(batch: ProducerBatch): Promise<RecordMetadata[]>
+    * @param {import('../../types/kafkajs').ProducerBatch} sendOptions - The record to send. The keys `acks`, `timeout`, and `compression` are not used, and should not be set, rather, they should be set in the global config.
+    * @returns {Promise<import("../../types/kafkajs").RecordMetadata[]>} Resolves with the record metadata for the messages.
+  */
+  async sendBatch(sendOptions) {
+    if (this.#state !== ProducerState.CONNECTED) {
+      throw new error.KafkaJSError("Cannot send without awaiting connect()", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    if (sendOptions === null || !(sendOptions instanceof Object)) {
+      throw new error.KafkaJSError("sendOptions must be set correctly", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    // Ignore all properties except topic and messages.
+    if (Object.hasOwn(sendOptions, "acks") || Object.hasOwn(sendOptions, "timeout") || Object.hasOwn(sendOptions, "compression")) {
+      throw new error.KafkaJSError("sendOptions must not contain acks, timeout, or compression", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    if (sendOptions.topicMessages !== null && !Array.isArray(sendOptions.topicMessages)) {
+      throw new error.KafkaJSError("sendOptions.topicMessages must be an array if set", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    if (!sendOptions.topicMessages || sendOptions.topicMessages.length === 0) {
+      return Promise.resolve([]);
+    }
+
+    // Internally, we just use send() because the batching is handled by librdkafka.
+    const sentPromises = [];
+
+    for (const topicMessage of sendOptions.topicMessages) {
+      sentPromises.push(this.send(topicMessage));
+    }
+
+    const records = await Promise.all(sentPromises);
+    return records.flat();
+  }
 }
 
 module.exports = { Producer }