Fixes #554 argument is out of bounds exception (#556)

SOHU-Co · Jan 12, 2017 · 9bcfb9d · 9bcfb9d
1 parent b9bef30
commit 9bcfb9d
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 0 deletions.
diff --git a/lib/client.js b/lib/client.js
@@ -22,6 +22,8 @@ var debug = require('debug')('kafka-node:Client');
 var validateConfig = require('./utils').validateConfig;
 var validateKafkaTopics = require('./utils').validateTopicNames;
 
+const MAX_INT32 = 2147483647;
+
 /**
  * Communicates with kafka brokers
  * Uses zookeeper to discover all the kafka brokers to connect to
@@ -445,6 +447,9 @@ Client.prototype.addTopics = function (topics, cb) {
 };
 
 Client.prototype.nextId = function () {
+  if (this.correlationId >= MAX_INT32) {
+    this.correlationId = 0;
+  }
   return this.correlationId++;
 };
 

diff --git a/test/test.consumerGroup.js b/test/test.consumerGroup.js
@@ -287,6 +287,31 @@ describe('ConsumerGroup', function () {
     });
   });
 
+  describe('Long running fetches', function () {
+    let consumerGroup;
+
+    beforeEach(function (done) {
+      consumerGroup = new ConsumerGroup({
+        host: host,
+        groupId: 'longFetchSimulation'
+      }, 'TestTopic');
+      consumerGroup.once('connect', done);
+    });
+
+    afterEach(function (done) {
+      consumerGroup.close(done);
+    });
+
+    it('should not throw out of bounds', function (done) {
+      should.doesNotThrow(function () {
+        consumerGroup.pause();
+        consumerGroup.client.correlationId = 2147483647;
+        consumerGroup.resume();
+        setImmediate(done);
+      });
+    });
+  });
+
   describe('Sending Heartbeats', function () {
     var consumerGroup, sandbox;