Add test for consumer.run().

Author: milindl

MIGRATION.md

@@ -178,6 +178,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).
+* A transactional producer (with a `transactionId`) set, can only send messages after calling `producer.transaction()`.
 
 ### Consumer
 

examples/kafkajs/eos.js

@@ -66,7 +66,9 @@ async function eosStart() {
                         {
                             topic,
                             partitions: [
-                                { partition, offset: message.offset },
+                                /* The message.offset indicates current offset, so we need to add 1 to it, since committed offset denotes
+                                 * the next offset to consume. */
+                                { partition, offset: message.offset + 1 },
                             ],
                         }
                     ],

lib/kafkajs/_consumer.js

@@ -449,6 +449,14 @@ class Consumer {
 
     while (this.#state === ConsumerState.CONNECTED) {
       const m = await this.#consumeSingle();
+
+      /* It's possible for the state to change as we await something.
+       * Unfortunately, we have no alternative but to break and drop the message, if it exists.
+       * TODO: fix this, possibly with a flag in disconnect that waits until the run loop winds down.
+       */
+      if (this.#state !== ConsumerState.CONNECTED)
+        break;
+
       if (!m) {
         continue;
       }
@@ -470,35 +478,57 @@ class Consumer {
           /* Don't pass this message on to the user if this topic partition was seeked to. */
           continue;
         }
+
+        /* It's possible for the state to change as we await something.
+         * Unfortunately, we have no alternative but to break and drop the message.
+         * TODO: fix this, possibly with a flag in disconnect that waits until the run loop winds down.
+         */
+        if (this.#state !== ConsumerState.CONNECTED)
+          break;
       }
 
       try {
         await config.eachMessage(
           this.#createPayload(m)
         )
+
+        /* It's possible for the state to change as we await something.
+         * Unfortunately, we have no alternative but to break without taking any action that the user might need.
+         */
+        if (this.#state !== ConsumerState.CONNECTED)
+          break;
       } catch (e) {
         /* It's not only possible, but expected that an error will be thrown by eachMessage.
          * This is especially true since the pattern of pause() followed by throwing an error
          * is encouraged. To meet the API contract, we seek one offset backward at this point (which
          * means seeking to the message offset). */
-        this.seek({
-          topic: m.topic,
-          partition: m.partition,
-          offset: m.offset,
-        });
+        if (this.#state === ConsumerState.CONNECTED)
+          this.seek({
+            topic: m.topic,
+            partition: m.partition,
+            offset: m.offset,
+          });
       }
 
       /* Force a immediate seek here. It's possible that there are no more messages to be passed to the user,
        * but the user seeked in the call to eachMessage, or else we encountered the error catch block.
        * In that case, the results of that seek will never be reflected unless we do this. */
-      if (this.#checkPendingSeeks)
+      if (this.#checkPendingSeeks && this.#state === ConsumerState.CONNECTED)
         await this.#seekInternal();
 
       /* TODO: another check we need to do here is to see how kafkaJS is handling
        * commits. Are they commmitting after a message is _processed_?
        * In that case we need to turn off librdkafka's auto-commit, and commit
        * inside this function.
        */
+
+      /* Yield for a bit to allow other scheduled tasks on the event loop to run.
+       * For instance, if disconnect() is called during/after we await eachMessage, and
+       * we don't await anything else after that, this loop will run despite needing to
+       * disconnect.
+       * It's better than any pending tasks be processed here, while we've processed one
+       * message completely, rather than between message processing. */
+      await new Promise((resolve) => setTimeout(resolve, 0));
     }
   }
 
@@ -573,7 +603,6 @@ class Consumer {
    */
   async #seekInternal(messageTopicPartition) {
     this.#checkPendingSeeks = false;
-
     const assignment = this.assignment();
     const offsetsToCommit = [];
     let invalidateMessage = false;

lib/kafkajs/_producer.js

@@ -169,8 +169,8 @@ class Producer {
       throw new error.KafkaJSError(`Ready callback called in invalid state ${this.#state}`, { code: error.ErrorCodes.ERR__STATE });
     }
 
-    const config = await this.#config();
-    if (Object.hasOwn(config, 'transactional.id') && this.#state !== ProducerState.INITIALIZED_TRANSACTIONS) {
+    const {globalConfig} = await this.#config();
+    if (Object.hasOwn(globalConfig, 'transactional.id') && this.#state !== ProducerState.INITIALIZED_TRANSACTIONS) {
       this.#state = ProducerState.INITIALIZING_TRANSACTIONS;
       this.#internalClient.initTransactions(5000 /* default: 5s */, this.#readyTransactions.bind(this));
       return;
@@ -450,7 +450,11 @@ class Producer {
 
       msgPromises.push(new Promise((resolve, reject) => {
         const opaque = { resolve, reject };
-        this.#internalClient.produce(sendOptions.topic, msg.partition, msg.value, msg.key, msg.timestamp, opaque, msg.headers);
+        try {
+          this.#internalClient.produce(sendOptions.topic, msg.partition, msg.value, msg.key, msg.timestamp, opaque, msg.headers);
+        } catch(err) {
+          reject(err);
+        }
       }));
 
     }