code: function(params, cb) {
      if (_.isString(params)) {
        params = { topic: params };
      }
  
      assert(params.topic);
      assert(cb);
      _closing = false;
  
      if (!params.maxWait) { params.maxWait = _options.maxWait; }
      if (!params.minBytes) { params.minBytes = _options.minBytes; }
  
      _initiate(params, function(err, topic) {
        if (err) { cb(err, null); return; }
        _poll(topic, null, cb);
      });
    }
docs: long poll kafka for data, will repeatedly poll on a topic / partition until
  endPoll is called. On calling poll an event emitter for the topic partition pair
  is returned, this should be used to receive events on the given channel
  
  params:
  topic        - the topic name, required
  partition    - the partition id, required
  offset       - the starting offset, if unspecified kafkaesque uses the latest commmited position
  maxWait      - the maximum poll wait time, if unspecified defaults to 5 seconds
  minBytes     - the minimum bytes that should be available before returning, if unspecified defaults to 50 bytes
  
  cb: callback of the form function(err, eventEmitter, commit):
  err          - error condtion
  eventEmitter - event emitter for this topic / partition
  commit       - commit function to be called on message processing to advance the kafka commit log
  events MUST be comitted in order to release the next message to the event emitter