kafka-node - npm Package Compare versions

Comparing version 0.2.29 to 0.2.30

lib/client.js

@@ -81,2 +81,9 @@ 'use strict';
     });
+    zk.once('disconnected', function () {
+        if (!zk.closed) {
+            zk.close();
+            self.connect();
+            self.emit('zkReconnect');
+        }
+    })
     zk.on('error', function (err) {
@@ -512,4 +519,10 @@ self.emit('error', err);
     socket.on('connect', function () {
+        var lastError = this.error;
         this.error = null;
-        self.emit('connect');
+        if (lastError) {
+            this.waitting = false;
+            self.emit('reconnect');
+        } else {
+            self.emit('connect');
+        }
     });
@@ -516,0 +529,0 @@ socket.on('error', function (err) {

lib/highLevelConsumer.js

@@ -51,3 +51,3 @@ 'use strict';
     this.rebalancing = false;
-    this.id = this.options.groupId + '_' + uuid.v4();
+    this.id = this.options.id || this.options.groupId + '_' + uuid.v4();
     this.payloads = this.buildPayloads(topics);
@@ -208,6 +208,18 @@ this.topicPayloads = this.buildTopicPayloads(topics);
 
-    this.client.zk.on('error', function (err) {
+    function attachZookeeperErrorListener() {
+        self.client.zk.on('error', function (err) {
         self.emit('error', err);
     });
+    }
 
+    attachZookeeperErrorListener();
+
+    this.client.on('zkReconnect', function () {
+        attachZookeeperErrorListener();
+
+        self.registerConsumer(function () {
+            rebalance();
+        });
+    });
+
     this.client.on('error', function (err) {
@@ -217,2 +229,6 @@ self.emit('error', err);
 
+    this.client.on('reconnect', function(lastError){
+        self.fetch();
+    })
+
     this.client.on('close', function () {
@@ -219,0 +235,0 @@ debug('close');

lib/protocol/protocol.js

@@ -369,5 +369,12 @@ 'use strict';
         .Int8(message.magic)
-        .Int8(message.attributes)
-        .Int32BE(message.key.length)
-        .string(message.key);
+        .Int8(message.attributes);
+
+    var key = message.key;
+    if (key) {
+        m.Int32BE(message.key.length);
+        m.string(message.key);
+    } else {
+        m.Int32BE(-1);
+    }
+
     var value = message.value;
@@ -374,0 +381,0 @@

lib/zookeeper.js

@@ -24,2 +24,5 @@ 'use strict';
     });
+    this.client.on('disconnected', function () {
+        that.emit('disconnected');
+    });
     this.client.connect();
@@ -138,3 +141,3 @@ };
                                             debug(e);
-                                            callback(new Error("Unable to assemble data"));
+                                            cbb(new Error("Unable to assemble data"));
                                         }
@@ -218,3 +221,3 @@ }
                     if (err) {
-                        that.emit('error', error);
+                        that.emit('error', err);
                         return;
@@ -221,0 +224,0 @@ }

package.json

 {
   "name": "kafka-node",
   "description": "node client for Apache kafka, only support kafka 0.8 and above",
-  "version": "0.2.29",
+  "version": "0.2.30",
   "main": "kafka.js",
@@ -6,0 +6,0 @@ "dependencies": {

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

151187

increased by0.59%

Lines of code

3452

increased by0.97%

No dependency changes