Move activation of dispatchers into event loop thread (#190)

milindl · web-flow · commit 7ec8cdaedc8d · 2024-12-12T19:22:54.000+05:30
* Reduce test flakiness

* Move activation of dispatchers to event loop thread

* Enable more tests on semaphore CI

* Update changelog and code comments

* Reject connection only after we're done with the entire process
diff --git a/.semaphore/semaphore.yml b/.semaphore/semaphore.yml
@@ -106,7 +106,7 @@ blocks:
             - '[[ -z $DOCKERHUB_APIKEY ]] || docker login --username $DOCKERHUB_USER --password $DOCKERHUB_APIKEY'
             - docker compose up -d && sleep 30
             - export NODE_OPTIONS='--max-old-space-size=1536'
-            - npx jest --forceExit --no-colors --ci test/promisified/admin/delete_groups.spec.js test/promisified/consumer/pause.spec.js
+            - npx jest --no-colors --ci test/promisified/
         - name: "ESLint"
           commands:
             - npx eslint lib/kafkajs
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,7 +4,12 @@ v1.0.0 is a feature release. It is supported for all usage.
 
 ## Enhancements
 
-1. Add support for an Admin API to fetch topic offsets(#156).
+1. Add support for an Admin API to fetch topic offsets (#156).
+
+## Fixes
+
+1. Fixes an issue where `uv_async_init` was being called off the event loop thread,
+   causing the node process to hang (#190).
 
 
 # confluent-kafka-javascript v0.6.1
@@ -29,7 +34,7 @@ v0.6.0 is a limited availability maintenance release. It is supported for all us
 
 ### Schema Registry
 
-1. Add AWS AssumeRole support to AWS KMS. You can now specify a role arn, and optional 
+1. Add AWS AssumeRole support to AWS KMS. You can now specify a role arn, and optional
 role session name and optional role external id.
 
 2. Ensure different key ids use different client instances.
diff --git a/lib/kafkajs/_admin.js b/lib/kafkajs/_admin.js
@@ -62,6 +62,13 @@ class Admin {
    */
   #connectPromiseFunc = null;
 
+  /**
+   * Stores the first error encountered while connecting (if any). This is what we
+   * want to reject with.
+   * @type {Error|null}
+   */
+  #connectionError = null;
+
   /**
    * The client name used by the admin client for logging - determined by librdkafka
    * using a combination of clientId and an integer.
@@ -169,8 +176,10 @@ class Admin {
    * @param {Error} err
    */
   #errorCb(err) {
-    if (this.#state === AdminState.CONNECTING) {
-      this.#connectPromiseFunc['reject'](err);
+    /* If we get an error in the middle of connecting, reject the promise later with this error. */
+    if (this.#state < AdminState.CONNECTED) {
+      if (!this.#connectionError)
+        this.#connectionError = err;
     } else {
       this.#logger.error(`Error: ${err.message}`, this.#createAdminBindingMessageMetadata());
     }
@@ -212,7 +221,9 @@ class Admin {
         this.#clientName = this.#internalClient.name;
         this.#logger.info("Admin client connected", this.#createAdminBindingMessageMetadata());
       } catch (err) {
-        reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        this.#state = AdminState.DISCONNECTED;
+        const rejectionError = this.#connectionError ? this.#connectionError : err;
+        reject(createKafkaJsErrorFromLibRdKafkaError(rejectionError));
       }
     });
   }
diff --git a/lib/kafkajs/_consumer.js b/lib/kafkajs/_consumer.js
@@ -65,6 +65,13 @@ class Consumer {
    */
   #connectPromiseFunc = {};
 
+  /**
+   * Stores the first error encountered while connecting (if any). This is what we
+   * want to reject with.
+   * @type {Error|null}
+   */
+  #connectionError = null;
+
   /**
    * state is the current state of the consumer.
    * @type {ConsumerState}
@@ -668,8 +675,10 @@ class Consumer {
    * @param {Error} err
    */
   #errorCb(err) {
-    if (this.#state === ConsumerState.CONNECTING) {
-      this.#connectPromiseFunc['reject'](err);
+    /* If we get an error in the middle of connecting, reject the promise later with this error. */
+    if (this.#state < ConsumerState.CONNECTED) {
+      if (!this.#connectionError)
+        this.#connectionError = err;
     } else {
       this.#logger.error(err, this.#createConsumerBindingMessageMetadata());
     }
@@ -1027,8 +1036,11 @@ class Consumer {
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };
       this.#internalClient.connect(null, (err) => {
-        if (err)
-          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        if (err) {
+          this.#state = ConsumerState.DISCONNECTED;
+          const rejectionError = this.#connectionError ? this.#connectionError : err;
+          reject(createKafkaJsErrorFromLibRdKafkaError(rejectionError));
+        }
       });
     });
   }
diff --git a/lib/kafkajs/_producer.js b/lib/kafkajs/_producer.js
@@ -58,6 +58,13 @@ class Producer {
    */
   #connectPromiseFunc = {};
 
+  /**
+   * Stores the first error encountered while connecting (if any). This is what we
+   * want to reject with.
+   * @type {Error|null}
+   */
+  #connectionError = null;
+
   /**
    * state is the current state of the producer.
    * @type {ProducerState}
@@ -330,11 +337,12 @@ class Producer {
    * @param {Error} err
    */
   #errorCb(err) {
-    if (this.#state === ProducerState.CONNECTING) {
-      this.#connectPromiseFunc["reject"](err);
-    } else {
-      this.#logger.error(err, this.#createProducerBindingMessageMetadata());
+    /* If we get an error in the middle of connecting, reject the promise later with this error. */
+    if (this.#state < ProducerState.CONNECTED) {
+      if (!this.#connectionError)
+        this.#connectionError = err;
     }
+    this.#logger.error(err, this.#createProducerBindingMessageMetadata());
   }
 
   /**
@@ -359,8 +367,11 @@ class Producer {
     return new Promise((resolve, reject) => {
       this.#connectPromiseFunc = { resolve, reject };
       this.#internalClient.connect(null, (err) => {
-        if (err)
-          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        if (err) {
+          this.#state = ProducerState.DISCONNECTED;
+          const rejectionError = this.#connectionError ? this.#connectionError : err;
+          reject(createKafkaJsErrorFromLibRdKafkaError(rejectionError));
+        }
       });
     });
   }
diff --git a/src/admin.cc b/src/admin.cc
@@ -56,11 +56,6 @@ Baton AdminClient::Connect() {
     return baton;
   }
 
-  // Activate the dispatchers before the connection, as some callbacks may run
-  // on the background thread.
-  // We will deactivate them if the connection fails.
-  ActivateDispatchers();
-
   std::string errstr;
   {
     scoped_shared_write_lock lock(m_connection_lock);
@@ -999,6 +994,14 @@ NAN_METHOD(AdminClient::NodeConnect) {
 
   AdminClient* client = ObjectWrap::Unwrap<AdminClient>(info.This());
 
+  // Activate the dispatchers before the connection, as some callbacks may run
+  // on the background thread.
+  // We will deactivate them if the connection fails.
+  // Because the Admin Client connect is synchronous, we can do this within
+  // AdminClient::Connect as well, but we do it here to keep the code similiar to
+  // the Producer and Consumer.
+  client->ActivateDispatchers();
+
   Baton b = client->Connect();
   // Let the JS library throw if we need to so the error can be more rich
   int error_code = static_cast<int>(b.err());
diff --git a/src/kafka-consumer.cc b/src/kafka-consumer.cc
@@ -1450,6 +1450,11 @@ NAN_METHOD(KafkaConsumer::NodeConnect) {
 
   KafkaConsumer* consumer = ObjectWrap::Unwrap<KafkaConsumer>(info.This());
 
+  // Activate the dispatchers before the connection, as some callbacks may run
+  // on the background thread.
+  // We will deactivate them if the connection fails.
+  consumer->ActivateDispatchers();
+
   Nan::Callback *callback = new Nan::Callback(info[0].As<v8::Function>());
   Nan::AsyncQueueWorker(new Workers::KafkaConsumerConnect(callback, consumer));
 
diff --git a/src/producer.cc b/src/producer.cc
@@ -729,6 +729,12 @@ NAN_METHOD(Producer::NodeConnect) {
   Nan::Callback *callback = new Nan::Callback(cb);
 
   Producer* producer = ObjectWrap::Unwrap<Producer>(info.This());
+
+  // Activate the dispatchers before the connection, as some callbacks may run
+  // on the background thread.
+  // We will deactivate them if the connection fails.
+  producer->ActivateDispatchers();
+
   Nan::AsyncQueueWorker(new Workers::ProducerConnect(callback, producer));
 
   info.GetReturnValue().Set(Nan::Null());
diff --git a/src/workers.cc b/src/workers.cc
@@ -199,11 +199,6 @@ ProducerConnect::ProducerConnect(Nan::Callback *callback, Producer* producer):
 ProducerConnect::~ProducerConnect() {}
 
 void ProducerConnect::Execute() {
-  // Activate the dispatchers before the connection, as some callbacks may run
-  // on the background thread.
-  // We will deactivate them if the connection fails.
-  producer->ActivateDispatchers();
-
   Baton b = producer->Connect();
 
   if (b.err() != RdKafka::ERR_NO_ERROR) {
@@ -558,11 +553,6 @@ KafkaConsumerConnect::KafkaConsumerConnect(Nan::Callback *callback,
 KafkaConsumerConnect::~KafkaConsumerConnect() {}
 
 void KafkaConsumerConnect::Execute() {
-  // Activate the dispatchers before the connection, as some callbacks may run
-  // on the background thread.
-  // We will deactivate them if the connection fails.
-  consumer->ActivateDispatchers();
-
   Baton b = consumer->Connect();
   // consumer->Wait();
 
diff --git a/test/promisified/admin/describe_groups.spec.js b/test/promisified/admin/describe_groups.spec.js
@@ -2,6 +2,7 @@ jest.setTimeout(30000);
 
 const {
     createConsumer,
+    createProducer,
     secureRandom,
     createTopic,
     waitFor,
@@ -11,18 +12,21 @@ const {
 const { ConsumerGroupStates, ErrorCodes, AclOperationTypes } = require('../../../lib').KafkaJS;
 
 describe('Admin > describeGroups', () => {
-    let topicName, groupId, consumer, admin;
+    let topicName, groupId, consumer, admin, groupInstanceId, producer;
 
     beforeEach(async () => {
         topicName = `test-topic-${secureRandom()}`;
         groupId = `consumer-group-id-${secureRandom()}`;
+        groupInstanceId = `consumer-group-instance-id-${secureRandom()}`;
+
+        producer = createProducer({});
 
         consumer = createConsumer({
             groupId,
             fromBeginning: true,
             clientId: 'test-client-id',
         }, {
-            'group.instance.id': 'test-instance-id',
+            'group.instance.id': groupInstanceId,
             'session.timeout.ms': 10000,
             'partition.assignment.strategy': 'roundrobin',
         });
@@ -33,6 +37,7 @@ describe('Admin > describeGroups', () => {
     });
 
     afterEach(async () => {
+        producer && (await producer.disconnect());
         consumer && (await consumer.disconnect());
         admin && (await admin.disconnect());
     });
@@ -61,9 +66,11 @@ describe('Admin > describeGroups', () => {
     });
 
     it('should describe consumer groups', async () => {
+        let messagesConsumed = 0;
+
         await consumer.connect();
         await consumer.subscribe({ topic: topicName });
-        await consumer.run({ eachMessage: async () => { } });
+        await consumer.run({ eachMessage: async () => { messagesConsumed++; } });
 
         await waitFor(() => consumer.assignment().length > 0, () => null, 1000);
 
@@ -92,7 +99,7 @@ describe('Admin > describeGroups', () => {
                         memberId: expect.any(String),
                         memberAssignment: null,
                         memberMetadata: null,
-                        groupInstanceId: 'test-instance-id',
+                        groupInstanceId: groupInstanceId,
                         assignment: {
                             topicPartitions:[
                                 expect.objectContaining({ topic: topicName, partition: 0 }),
@@ -104,12 +111,22 @@ describe('Admin > describeGroups', () => {
             })
         );
 
-        // Disconnect the consumer to make the group EMPTY.
+        // Produce some messages so that the consumer can commit them, and hence
+        // the group doesn't become DEAD.
+        await producer.connect();
+        await producer.send({
+            topic: topicName,
+            messages: [{ key: 'key', value: 'value' }],
+        });
+
+        await waitFor(() => messagesConsumed > 0, () => null, 1000);
+
+        // Disconnect the consumer to make the group EMPTY and commit offsets.
         await consumer.disconnect();
         consumer = null;
 
         // Wait so that session.timeout.ms expires and the group becomes EMPTY.
-        await sleep(12000);
+        await sleep(10500);
 
         // Don't include authorized operations this time.
         describeGroupsResult = await admin.describeGroups([groupId]);
@@ -119,9 +136,9 @@ describe('Admin > describeGroups', () => {
                 groupId,
                 protocol: '',
                 partitionAssignor: '',
-                isSimpleConsumerGroup: false,
-                protocolType: 'consumer',
                 state: ConsumerGroupStates.EMPTY,
+                protocolType: 'consumer',
+                isSimpleConsumerGroup: false,
                 coordinator: expect.objectContaining({
                     id: expect.any(Number),
                     host: expect.any(String),
@@ -130,6 +147,7 @@ describe('Admin > describeGroups', () => {
                 members: [],
             })
         );
+
         expect(describeGroupsResult.groups[0].authorizedOperations).toBeUndefined();
     });
 });
diff --git a/test/promisified/consumer/consumeMessages.spec.js b/test/promisified/consumer/consumeMessages.spec.js
diff --git a/test/promisified/consumer/consumerCacheTests.spec.js b/test/promisified/consumer/consumerCacheTests.spec.js
diff --git a/test/promisified/oauthbearer_cb.spec.js b/test/promisified/oauthbearer_cb.spec.js
diff --git a/test/promisified/testhelpers.js b/test/promisified/testhelpers.js
