title: Read specific messages using Apache Kafka and NodeJS
body: <p>I want to build an API with NodeJS and Kafka, which can take an offset and a topic as an input and output the first 10 messages starting from the offset. I tried this approach with <a href="https://www.npmjs.com/package/no-kafka" rel="nofollow">No-Kafka</a> and <a href="https://www.npmjs.com/package/kafka-node" rel="nofollow">Kafka-Node</a>. </p>
  
  <p>The consumer API provided by them allows consuming messages from a particular offset. I want to stop consuming the messages once I have read around 10 messages. But both the API calls will continue to fetch the messages till the last message. How can I stop doing that?</p>
  
  <p>Here is my <strong>EDITED FULL CODE</strong></p>
  
  <pre><code>var Kafka = require('no-kafka');
  var express = require("express");
  var app = express();
  
  var producer = new Kafka.Producer();
  producer.init().then(function() {
    console.log("Producer Ready");
  });
  
  var consumer = new Kafka.SimpleConsumer();
  consumer.init().then(function() {
    console.log("Consumer Ready");
  });
  
  app.get('/produce/:topic/:msg', function(req, res) {
    producer.send({
      topic: req.params.topic,
      partition: 0,
      message: {
        value: req.params.msg
      }
    });
    res.send("Added: " + req.params.msg + " to topic: " + req.params.topic);
  });
  
  app.get('/consume/:topic/:off', function(req, res) {
    console.log("Request for topic: " + req.params.topic + " Offset: " + req.params.off);
    consumer.subscribe(req.params.topic, 0, {
      offset: req.params.off,
      maxBytes: 1000
    }, function(messageSet, topic, partition) {
      var msg = "";
      var size = messageSet.length;
      //console.log(messageSet);
      messageSet.some(function(m) {
        msg += m.message.value.toString('utf8') + " ";
        if (parseInt(m.offset, 10) &gt; parseInt(req.params.off, 10) + 10) {
          return true;
        }
      });
      res.send("Thank you " + size + "  " + req.params.off + "  " + msg);
    });
  });
  
  app.listen(process.env.PORT);
  </code></pre>
  
  <p>Any response in this regard is appreciated.</p>