<?xml version="1.0" encoding="UTF-8"?>

 <project xmlns="http://maven.apache.org/POM/4.0.0"

          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

     <modelVersion>4.0.0</modelVersion>

     <groupId>com.ht</groupId>

     <artifactId>kafkatest</artifactId>

     <version>1.0-SNAPSHOT</version>

     <build>

         <plugins>

             <plugin>

                 <groupId>org.apache.maven.plugins</groupId>

                 <artifactId>maven-compiler-plugin</artifactId>

                 <configuration>

                     <source>1.7</source>

                     <target>1.7</target>

                 </configuration>

             </plugin>

         </plugins>

     </build>

     <dependencies>

         <dependency>

             <groupId>org.apache.kafka</groupId>

             <artifactId>kafka-clients</artifactId>

             <version>0.10.0.0</version>

         </dependency>

     </dependencies>

 </project>

 import org.apache.kafka.clients.CommonClientConfigs;

 import org.apache.kafka.clients.consumer.ConsumerRecord;

 import org.apache.kafka.clients.consumer.ConsumerRecords;

 import org.apache.kafka.clients.consumer.KafkaConsumer;

 import java.util.Collections;

 import java.util.Properties;

 import static org.apache.kafka.clients.consumer.ConsumerConfig.*;

 /**

  * @author sunzq

  * @since 2017/8/29

  */

 public class Application {

     public static void main(String[] args) {

         Properties props = new Properties();

         props.put(BOOTSTRAP_SERVERS_CONFIG, "node1:6667,node2:6667,node3:6667,node4:6667");

         props.put(ENABLE_AUTO_COMMIT_CONFIG, "true");

         props.put(GROUP_ID_CONFIG, "test08291103");

 //      props.put(ConsumerConfig.CLIENT_ID_CONFIG, "test0829");

         props.put(AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

         props.put(AUTO_OFFSET_RESET_CONFIG, "earliest");

         props.put(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

         props.put(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");

         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

         // topic name: test9

         consumer.subscribe(Collections.singleton("test9"));

         while (true) {

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

             for (ConsumerRecord<String, String> record : records)

                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

         }

     }

 }