Fix blocked ConsumerGroup fetch loop when fetch request yields an error

closes #1179
SOHU-Co · Feb 22, 2019 · 8c0c22f · 8c0c22f
1 parent 7c857d1
commit 8c0c22f
Show file tree

Hide file tree

Showing 2 changed files with 75 additions and 1 deletion.
diff --git a/lib/consumerGroup.js b/lib/consumerGroup.js
@@ -815,7 +815,12 @@ ConsumerGroup.prototype.fetch = function () {
     this.topicPayloads,
     this.options.fetchMaxWaitMs,
     this.options.fetchMinBytes,
-    this.options.maxTickMessages
+    this.options.maxTickMessages,
+    error => {
+      if (error) {
+        this._resetFetchState();
+      }
+    }
   );
 };
 

diff --git a/test/test.consumerGroup.js b/test/test.consumerGroup.js
@@ -12,6 +12,7 @@ const uuid = require('uuid');
 const async = require('async');
 const BrokerWrapper = require('../lib/wrapper/BrokerWrapper');
 const FakeSocket = require('./mocks/mockSocket');
+const { BrokerNotAvailableError } = require('../lib/errors');
 
 describe('ConsumerGroup', function () {
   describe('#constructor', function () {
@@ -1095,6 +1096,74 @@ describe('ConsumerGroup', function () {
     });
   });
 
+  describe('#fetch', function () {
+    let consumerGroup;
+
+    afterEach(function (done) {
+      consumerGroup.close(done);
+    });
+
+    it('should reset fetch pending state if fetch request fails', function () {
+      const topic = uuid.v4();
+      consumerGroup = new ConsumerGroup(
+        {
+          connectOnReady: false,
+          groupId: uuid.v4(),
+          autoCommit: false
+        },
+        [topic]
+      );
+
+      consumerGroup.ready = true;
+      consumerGroup.paused = false;
+      consumerGroup._isFetchPending = false;
+
+      const clientMock = sinon.mock(consumerGroup.client);
+
+      clientMock
+        .expects('sendFetchRequest')
+        .once()
+        .yields(new BrokerNotAvailableError('Test Error'));
+
+      const cgMock = sinon.mock(consumerGroup);
+      cgMock.expects('_resetFetchState').once();
+
+      consumerGroup.fetch();
+      clientMock.verify();
+      cgMock.verify();
+    });
+
+    it('should not reset fetch pending state if fetch request was successful', function () {
+      const topic = uuid.v4();
+      consumerGroup = new ConsumerGroup(
+        {
+          connectOnReady: false,
+          groupId: uuid.v4(),
+          autoCommit: false
+        },
+        [topic]
+      );
+
+      consumerGroup.ready = true;
+      consumerGroup.paused = false;
+      consumerGroup._isFetchPending = false;
+
+      const clientMock = sinon.mock(consumerGroup.client);
+
+      clientMock
+        .expects('sendFetchRequest')
+        .once()
+        .yields(null);
+
+      const cgMock = sinon.mock(consumerGroup);
+      cgMock.expects('_resetFetchState').never();
+
+      consumerGroup.fetch();
+      clientMock.verify();
+      cgMock.verify();
+    });
+  });
+
   describe('#removeTopics', function () {
     let topic, newTopic, testMessage, consumerGroup;