Fix intermittent test failures

SOHU-Co · Jul 8, 2017 · 88b99f8 · 88b99f8
1 parent 7601f6e
commit 88b99f8
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 3 deletions.
diff --git a/lib/consumerGroup.js b/lib/consumerGroup.js
@@ -106,6 +106,7 @@ function ConsumerGroup (memberOptions, topics) {
 
   if (this.options.migrateHLC) {
     if (this.client instanceof KafkaClient) {
+      this.client.close(_.noop);
       throw new Error('KafkaClient cannot be used to migrate from Zookeeper use Client instead');
     }
 
@@ -414,6 +415,11 @@ ConsumerGroup.prototype.connect = function () {
     return;
   }
 
+  if (this.closed) {
+    logger.warn('Connect ignored. Consumer closed.');
+    return;
+  }
+
   logger.debug('Connecting %s', this.client.clientId);
   var self = this;
 
@@ -600,7 +606,13 @@ ConsumerGroup.prototype.close = function (force, cb) {
         self.client.close(callback);
       }
     ],
-    cb
+    function (error) {
+      if (error) {
+        return cb(error);
+      }
+      self.closed = true;
+      cb(null);
+    }
   );
 };
 

diff --git a/test/test.baseProducer.js b/test/test.baseProducer.js
@@ -10,8 +10,8 @@ const async = require('async');
 const should = require('should');
 
 describe('BaseProducer', function () {
-  const KAFKA_HOST = 'localhost:9092';
   describe('encoding and decoding key attribute', function () {
+    const KAFKA_HOST = 'localhost:9092';
     let consumerGroup, topic, producer;
     beforeEach(function (done) {
       topic = uuid.v4();

diff --git a/test/test.consumerGroup.js b/test/test.consumerGroup.js
@@ -730,7 +730,7 @@ describe('ConsumerGroup', function () {
         new ConsumerGroup({
           kafkaHost: 'localhost:9092',
           migrateHLC: true
-        });
+        }, 'TestTopic');
       });
     });
   });