java_consumer.java

public static void main(String[] args) {
  KafkaConsumer consumer = new KafkaConsumer(getProperties());

  consumer.subscribe(Collections.singletonList("bids"));

  System.out.println("Starting polling from Java bids consumer:");

  while(true) {
      ConsumerRecords<String, String> records = consumer.poll(100);

      for (ConsumerRecord<String, String> record : records) {
          String message = String.format("key: %s, payload: %s", record.key(), record.value());
          System.out.println(message);
      }

      consumer.commitSync();
  }
}

private static Properties getProperties() {
  Properties props = new Properties();
  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", "HelloConsumer");
  props.put("key.deserializer",
          "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer",
          "org.apache.kafka.common.serialization.StringDeserializer");

  return props;
}