Connect directly to kafka brokers using new client closes #666

README.md

@@ -18,6 +18,7 @@ Kafka-node is a Node.js client with Zookeeper integration for Apache Kafka 0.8.1
 - [Features](#features)
 - [Install Kafka](#install-kafka)
 - [API](#api)
+  - [KafkaClient](#kafkaclient)
   - [Client](#client)
   - [Producer](#producer)
   - [HighLevelProducer](#highlevelproducer)
@@ -47,11 +48,37 @@ Kafka-node is a Node.js client with Zookeeper integration for Apache Kafka 0.8.1
 * Manage topic Offsets
 * SSL connections to brokers (Kafka 0.9+)
 * Consumer Groups managed by Kafka coordinator (Kafka 0.9+)
+* Connect directly to brokers (Kafka 0.9+)
 
 # Install Kafka
 Follow the [instructions](http://kafka.apache.org/documentation.html#quickstart) on the Kafka wiki to build Kafka 0.8 and get a test broker up and running.
 
 # API
+
+## KafkaClient
+
+New KafkaClient connects directly to Kafka brokers instead of connecting to zookeeper for broker discovery.
+
+### New Features
+
+* Kafka **ONLY** no zookeeper
+* Added request timeout
+* Added connection timeout and retry
+
+### Notable differences
+
+* Constructor accepts an single options object (see below)
+* Unlike the original `Client` `KafkaClient` will not emit socket errors it will do it's best to recover and only emit errors when it has exhausted it's recovery attempts
+* `ready` event is only emitted after successful connection to a broker and metadata request to that broker
+* `Client` uses zookeeper to discover the SSL kafka host/port since we connect directly to the broker this host/port for SSL need to be correct
+
+### Options
+* `kafkaHost` : A string of kafka broker/host combination delimited by comma for example: `kafka-1.us-east-1.myapp.com:9093,kafka-2.us-east-1.myapp.com:9093,kafka-3.us-east-1.myapp.com:9093` default: `localhost:9092`.
+* `connectTimeout` : in ms it takes to wait for a successful connection before moving to the next host default: `10000`
+* `requestTimeout` : in ms for a kafka request to timeout default: `30000`
+* `autoConnect` : automatically connect when KafkaClient is instantiated otherwise you need to manually call `connect` default: `true`
+* `connectRetryOptions` : object hash that applies to the initial connection. see [retry](https://www.npmjs.com/package/retry) module for these options.
+
 ## Client
 ### Client(connectionString, clientId, [zkOptions], [noAckBatchOptions], [sslOptions])
 * `connectionString`: Zookeeper connection string, default `localhost:2181/`
@@ -565,7 +592,8 @@ API is very similar to `HighLevelConsumer` since it extends directly from HLC so
 
 ```js
 var options = {
-  host: 'zookeeper:2181',
+  host: 'zookeeper:2181',  // zookeeper host omit if connecting directly to broker (see kafkaHost below)
+  kafkaHost: 'broker:9092', // connect directly to kafka broker (instantiates a KafkaClient)
   zk : undefined,   // put client zk settings if you need them (see Client)
   batch: undefined, // put client batch settings if you need them (see Client)
   ssl: true, // optional (defaults to false) or tls options hash

kafka.js

@@ -4,6 +4,7 @@ exports.ConsumerGroup = require('./lib/consumerGroup');
 exports.Consumer = require('./lib/consumer');
 exports.Producer = require('./lib/producer');
 exports.Client = require('./lib/client');
+exports.KafkaClient = require('./lib/kafkaClient');
 exports.Offset = require('./lib/offset');
 exports.KeyedMessage = require('./lib/protocol').KeyedMessage;
 exports.DefaultPartitioner = require('./lib/partitioner').DefaultPartitioner;

lib/client.js

@@ -78,21 +78,19 @@ var Client = function (connectionString, clientId, zkOptions, noAckBatchOptions,
 util.inherits(Client, events.EventEmitter);
 
 Client.prototype.connect = function () {
-  var zk = this.zk = new Zookeeper(this.connectionString, this.zkOptions);
+  var zk = (this.zk = new Zookeeper(this.connectionString, this.zkOptions));
   var self = this;
   zk.once('init', function (brokers) {
     try {
       self.ready = true;
       self.brokerMetadata = brokers;
       self.setupBrokerProfiles(brokers);
-      Object
-          .keys(self.brokerProfiles)
-          .some(function (key, index) {
-            var broker = self.brokerProfiles[key];
-            self.setupBroker(broker.host, broker.port, false, self.brokers);
-            // Only connect one broker
-            return !index;
-          });
+      Object.keys(self.brokerProfiles).some(function (key, index) {
+        var broker = self.brokerProfiles[key];
+        self.setupBroker(broker.host, broker.port, false, self.brokers);
+        // Only connect one broker
+        return !index;
+      });
       self.emit('ready');
     } catch (error) {
       self.ready = false;
@@ -238,7 +236,7 @@ Client.prototype.sendProduceRequest = function (payloads, requireAcks, ackTimeou
     var innerSet = encodeMessageSet(payload.messages);
     codec.encode(innerSet, function (err, message) {
       if (err) return cb(err);
-      payload.messages = [ new Message(0, attributes, '', message) ];
+      payload.messages = [new Message(0, attributes, '', message)];
       cb();
     });
   }
@@ -274,6 +272,8 @@ Client.prototype.sendOffsetRequest = function (payloads, cb) {
   this.send(payloads, encoder, decoder, cb);
 };
 
+Client.prototype.refreshBrokerMetadata = function () {};
+
 Client.prototype.sendGroupRequest = function (encode, decode, requestArgs) {
   requestArgs = _.values(requestArgs);
   var cb = requestArgs.pop();
@@ -285,6 +285,7 @@ Client.prototype.sendGroupRequest = function (encode, decode, requestArgs) {
   var broker = this.brokerForLeader(this.coordinatorId);
 
   if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
+    this.refreshBrokerMetadata();
     return cb(new errors.BrokerNotAvailableError('Broker not available'));
   }
 
@@ -464,11 +465,13 @@ Client.prototype.refreshBrokers = function () {
   function closeDeadBrokers (brokers) {
     var deadBrokerKeys = _.difference(Object.keys(brokers), validBrokers);
     if (deadBrokerKeys.length) {
-      self.closeBrokers(deadBrokerKeys.map(function (key) {
-        var broker = brokers[key];
-        delete brokers[key];
-        return broker;
-      }));
+      self.closeBrokers(
+        deadBrokerKeys.map(function (key) {
+          var broker = brokers[key];
+          delete brokers[key];
+          return broker;
+        })
+      );
     }
   }
 
@@ -511,7 +514,9 @@ Client.prototype.send = function (payloads, encoder, decoder, cb) {
     return;
   }
   if (payloads[1].length) {
-    var topicNames = payloads[1].map(function (p) { return p.topic; });
+    var topicNames = payloads[1].map(function (p) {
+      return p.topic;
+    });
     this.loadMetadataForTopics(topicNames, function (err, resp) {
       if (err) {
         return cb(err);
@@ -526,6 +531,7 @@ Client.prototype.send = function (payloads, encoder, decoder, cb) {
       // check payloads again
       payloads = self.checkMetadatas(_payloads);
       if (payloads[1].length) {
+        this.refreshBrokerMetadata();
         return cb(new errors.BrokerNotAvailableError('Could not find the leader'));
       }
 
@@ -548,6 +554,7 @@ Client.prototype.sendToBroker = function (payloads, encoder, decoder, cb) {
     var request = encoder(this.clientId, correlationId, payloads[leader]);
     var broker = this.brokerForLeader(leader, longpolling);
     if (!broker || !broker.socket || broker.socket.error || broker.socket.closing || broker.socket.destroyed) {
+      this.refreshBrokerMetadata();
       return cb(new errors.BrokerNotAvailableError('Could not find the leader'), payloads[leader]);
     }
 
@@ -567,21 +574,23 @@ Client.prototype.sendToBroker = function (payloads, encoder, decoder, cb) {
 };
 
 Client.prototype.checkMetadatas = function (payloads) {
-  if (_.isEmpty(this.topicMetadata)) return [ [], payloads ];
+  if (_.isEmpty(this.topicMetadata)) return [[], payloads];
   // out: [ [metadata exists], [metadata not exists] ]
-  var out = [ [], [] ];
-  payloads.forEach(function (p) {
-    if (this.hasMetadata(p.topic, p.partition)) out[0].push(p);
-    else out[1].push(p);
-  }.bind(this));
+  var out = [[], []];
+  payloads.forEach(
+    function (p) {
+      if (this.hasMetadata(p.topic, p.partition)) out[0].push(p);
+      else out[1].push(p);
+    }.bind(this)
+  );
   return out;
 };
 
 Client.prototype.hasMetadata = function (topic, partition) {
   var brokerMetadata = this.brokerMetadata;
   var leader = this.leaderByPartition(topic, partition);
 
-  return (leader !== undefined) && brokerMetadata[leader];
+  return leader !== undefined && brokerMetadata[leader];
 };
 
 Client.prototype.updateMetadatas = function (metadatas) {
@@ -598,19 +607,24 @@ Client.prototype.updateMetadatas = function (metadatas) {
 };
 
 Client.prototype.removeTopicMetadata = function (topics, cb) {
-  topics.forEach(function (t) {
-    if (this.topicMetadata[t]) delete this.topicMetadata[t];
-  }.bind(this));
+  topics.forEach(
+    function (t) {
+      if (this.topicMetadata[t]) delete this.topicMetadata[t];
+    }.bind(this)
+  );
   cb(null, topics.length);
 };
 
 Client.prototype.payloadsByLeader = function (payloads) {
-  return payloads.reduce(function (out, p) {
-    var leader = this.leaderByPartition(p.topic, p.partition);
-    out[leader] = out[leader] || [];
-    out[leader].push(p);
-    return out;
-  }.bind(this), {});
+  return payloads.reduce(
+    function (out, p) {
+      var leader = this.leaderByPartition(p.topic, p.partition);
+      out[leader] = out[leader] || [];
+      out[leader].push(p);
+      return out;
+    }.bind(this),
+    {}
+  );
 };
 
 Client.prototype.leaderByPartition = function (topic, partition) {
@@ -633,7 +647,7 @@ Client.prototype.brokerForLeader = function (leader, longpolling) {
     }
   }
 
-  var broker = _.find(this.brokerProfiles, {id: leader});
+  var broker = _.find(this.brokerProfiles, { id: leader });
 
   if (!broker) {
     return;
@@ -738,15 +752,19 @@ Client.prototype.handleReceivedData = function (socket) {
     var decoder = handlers[0];
     var cb = handlers[1];
     var result = decoder(resp);
-    (result instanceof Error)
-      ? cb.call(this, result)
-      : cb.call(this, null, result);
+    result instanceof Error ? cb.call(this, result) : cb.call(this, null, result);
     buffer.consume(size);
     if (socket.longpolling) socket.waiting = false;
-  } else { return; }
+  } else {
+    return;
+  }
 
   if (socket.buffer.length) {
-    setImmediate(function () { this.handleReceivedData(socket); }.bind(this));
+    setImmediate(
+      function () {
+        this.handleReceivedData(socket);
+      }.bind(this)
+    );
   }
 };