0

I've setted up an kafka-consumer on my servlet. I read from my kafka topic and want to print the latest value using the doGet-method of the servlet.

But if I send the doGet I just get a "null" back...

I'm not sure, but I found another post, where somebody mentioned, that this might be because there is no connection to the zookeeper servers. So I tried to fix this adding the zookeeper.connection to the properties of the consumer; even if it's outdated.

Hopefully someone could give me advice about this.

Code of my Servlet:

public class KafkaServlet extends HttpServlet implements Runnable {

public String kafkaMessage;
private Properties props;
private KafkaConsumer<String, String> consumer;
Thread Trans;

/**
*   setting up the properties and other consumer using the constructor of KafkaServlet
*/
public KafkaServlet() {
    props = new Properties();
    props.put("bootstrap.servers", "localhost:9092"); //replace 'localhost' with the actual IP to get it working.
    props.put("zookeeper.connect", "localhost:2181");
    props.put("auto.offset.reset", "earliest");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList("kafkaTopic"));
}

/**
*   setup the thread and run it using the init-method
*/
@Override
public void init (ServletConfig config) throws ServletException {
    super.init(config);
    Trans = new Thread(this);
    Trans.setPriority(Thread.MIN_PRIORITY);
    Trans.start();
}

/**
*   implement the doPost-Method if we want to use it 
*/
public void doPost(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {
    //empty -- nothing to post here
}

/**
*   doGet will get print out our returned message from kafka
*/
@Override
public void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {

res.setContentType("text/html");
PrintWriter out = res.getWriter();
String title = "Reading Parameters from kafkaTopic";

out.println("<HTML>" +
            "<BODY>\n" +
            "<H1 ALIGN=CENTER>" + title + "</H1>\n" +
            "<UL>\n" +
            "  <LI>MESSAGE: "
            + kafkaMessage + "\n" +
            "</UL>\n" +
            "</BODY></HTML>");
}

/**
*   thread, which grabs the messages from kafka and stores the latest one in "kafkaMessage"
*/
@Override
public void run() {
    try {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                //KafkaMessage.setMessage(record.value());
                kafkaMessage = record.value();
            }
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        consumer.close();
    }
}}

EDIT: Latest logfiles from "localhost.log":

19-Dec-2016 13:42:42.773 INFO [localhost-startStop-2] org.apache.catalina.core.ApplicationContext.log SessionListener: contextDestroyed() 19-Dec-2016 13:42:42.774 INFO [localhost-startStop-2] org.apache.catalina.core.ApplicationContext.log ContextListener: contextDestroyed() 19-Dec-2016 13:42:54.742 INFO [localhost-startStop-1] org.apache.catalina.core.ApplicationContext.log ContextListener: contextInitialized() 19-Dec-2016 13:42:54.745 INFO [localhost-startStop-1] org.apache.catalina.core.ApplicationContext.log SessionListener: contextInitialized()

2
  • 1
    It looks like you are trying to build an HTTP Proxy. You might want to look at the code for the existing REST Proxy from Confluent. github.com/confluentinc/kafka-rest Commented Dec 19, 2016 at 16:30
  • Thanks, I had a look into the sourcecode, but there were no significant differences for the config and setting up the consumer. But I figured the problem out, by replacing the "localhost" on the "bootstrap.servers" with my actual IP of my VM. Commented Dec 20, 2016 at 8:11

1 Answer 1

0

I figured it out: the problem, why I did not receive any messages was the line props.put("bootstrap.servers", "localhost:9092");.

I had to change the "localhost" to my actual IP of my VM. This is kinda strange, because every other property I configured (on apache storm; kafka; another servlet which contains an kafka producer) was working well with "localhost".

1
  • 1
    It's recommended that Kafka brokers run on standalone hardware or VMs so you probably want to run your Proxy on a separate machine when you get to production. Commented Dec 21, 2016 at 17:17

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.