broker is not started

src/main/java/edu/my/example/kafka/Consumer.java

@@ -0,0 +1,26 @@
+package edu.my.example.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.CountDownLatch;
+
+@Component
+public class Consumer {
+    private static final Logger log = LoggerFactory.getLogger(Consumer.class);
+    private CountDownLatch latch = new CountDownLatch(1);
+
+    public CountDownLatch getLatch() {
+        return latch;
+    }
+
+    @KafkaListener(topics = "${kafka.topic.boot}")
+    public void recevied(ConsumerRecord<?,?> consumerRecord){
+        log.info("received payload='{}'", consumerRecord.toString());
+        latch.countDown();
+    }
+}

src/main/java/edu/my/example/kafka/Sender.java

@@ -0,0 +1,20 @@
+package edu.my.example.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.stereotype.Component;
+
+@Component
+public class Sender {
+    private static final Logger log = LoggerFactory.getLogger(Sender.class);
+
+    @Autowired
+    private KafkaTemplate<String,String> kafkaTemplate;
+
+    public void send(String topic, String payload){
+        log.info("sending payload='{}' to topic ='{}'",payload,topic);
+        kafkaTemplate.send(topic,payload);
+    }
+}

src/main/resources/application.yaml

@@ -1,4 +1,9 @@
+spring:
+  kafka:
+    consumer:
+      auto-offset-reset: earliest
+      group-id: boot
+
 kafka:
-  bootstrap-servers: localhost:9092
   topic:
-    helloworld: helloworld.t
+    boot: boot.t

src/test/java/edu/my/example/kafka/SpringKafkaApplicationTest.java

@@ -0,0 +1,37 @@
+package edu.my.example.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.ClassRule;
+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.kafka.test.rule.KafkaEmbedded;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class SpringKafkaApplicationTest {
+
+  private static String BOOT_TOPIC = "boot.t";
+
+  @Autowired
+  private Sender sender;
+
+  @Autowired
+  private Consumer receiver;
+
+  @ClassRule
+  public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, BOOT_TOPIC);
+
+  @Test
+  public void testReceive() throws Exception {
+    sender.send(BOOT_TOPIC, "Hello Boot!");
+
+    receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
+    assertThat(receiver.getLatch().getCount()).isEqualTo(0);
+  }
+}