kafka support

@@ -22,6 +22,7 @@
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 		<java.version>1.8</java.version>
 		<java.version>1.8</java.version>
+		<spring-kafka.version>1.1.2.RELEASE</spring-kafka.version>
 	</properties>
 	</properties>
 
 
 	<dependencies>
 	<dependencies>
@@ -31,6 +32,12 @@
 		</dependency>
 		</dependency>
 
 
 		<dependency>
 		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>${spring-kafka.version}</version>
+		</dependency>
+
+		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 			<scope>test</scope>

+package com.yoho.ops.kafka;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+
+public class Receiver {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);
+
+  private CountDownLatch latch = new CountDownLatch(1);
+
+  @KafkaListener(topics = "${kafka.topic.helloworld}")
+  public void receive(String message) {
+    LOGGER.info("received message='{}'", message);
+    latch.countDown();
+  }
+
+  public CountDownLatch getLatch() {
+    return latch;
+  }
+}

+package com.yoho.ops.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+@Configuration
+@EnableKafka
+public class ReceiverConfig {
+
+  @Value("${kafka.servers.bootstrap}")
+  private String bootstrapServers;
+
+  @Bean
+  public Map<String, Object> consumerConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    // list of host:port pairs used for establishing the initial connections
+    // to the Kakfa cluster
+    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    // consumer groups allow a pool of processes to divide the work of
+    // consuming and processing records
+    props.put(ConsumerConfig.GROUP_ID_CONFIG, "helloworld");
+
+    return props;
+  }
+
+  @Bean
+  public ConsumerFactory<String, String> consumerFactory() {
+    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
+  }
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+    ConcurrentKafkaListenerContainerFactory<String, String> factory =
+        new ConcurrentKafkaListenerContainerFactory<>();
+    factory.setConsumerFactory(consumerFactory());
+
+    return factory;
+  }
+
+  @Bean
+  public Receiver receiver() {
+    return new Receiver();
+  }
+}

+package com.yoho.ops.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+public class Sender {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(Sender.class);
+
+  @Autowired
+  private KafkaTemplate<String, String> kafkaTemplate;
+
+  public void send(String topic, String message) {
+    // the KafkaTemplate provides asynchronous send methods returning a
+    // Future
+    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
+
+    // you can register a callback with the listener to receive the result
+    // of the send asynchronously
+    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
+
+      @Override
+      public void onSuccess(SendResult<String, String> result) {
+        LOGGER.info("sent message='{}' with offset={}", message,
+            result.getRecordMetadata().offset());
+      }
+
+      @Override
+      public void onFailure(Throwable ex) {
+        LOGGER.error("unable to send message='{}'", message, ex);
+      }
+    });
+
+    // alternatively, to block the sending thread, to await the result,
+    // invoke the future's get() method
+  }
+}

+package com.yoho.ops.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+public class SenderConfig {
+
+  @Value("${kafka.servers.bootstrap}")
+  private String bootstrapServers;
+
+  @Bean
+  public Map<String, Object> producerConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    // list of host:port pairs used for establishing the initial connections
+    // to the Kakfa cluster
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    // value to block, after which it will throw a TimeoutException
+    props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000);
+
+    return props;
+  }
+
+  @Bean
+  public ProducerFactory<String, String> producerFactory() {
+    return new DefaultKafkaProducerFactory<>(producerConfigs());
+  }
+
+  @Bean
+  public KafkaTemplate<String, String> kafkaTemplate() {
+    return new KafkaTemplate<>(producerFactory());
+  }
+
+  @Bean
+  public Sender sender() {
+    return new Sender();
+  }
+}

+kafka.servers.bootstrap=192.168.103.30:9092
+kafka.topic.helloworld=helloworld.t

+package com.yoho.ops.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class SpringKafkaApplicationTests {
+
+  @Autowired
+  private Sender sender;
+
+  @Autowired
+  private Receiver receiver;
+
+  @Test
+  public void testReceive() throws Exception {
+    sender.send("helloworld.t", "Hello Spring Kafka!");
+
+    Thread.sleep(100000);
+  }
+}