update projedt name + send method

Author: code-not-found

.travis.yml

@@ -18,7 +18,7 @@ env:
 - PROJECT_DIR=spring-kafka-boot
 - PROJECT_DIR=spring-kafka-helloworld
 - PROJECT_DIR=spring-kafka-json
-- PROJECT_DIR=spring-kafka-multiple-consumers
+- PROJECT_DIR=spring-kafka-multiple-topics
 - PROJECT_DIR=spring-kafka-spring-integration-helloworld
 - PROJECT_DIR=spring-kafka-test-embedded
 

spring-kafka-multiple-consumers/src/main/resources/application.yml

@@ -0,0 +1,7 @@
+# spring-kafka-multiple-topics
+
+[![Quality Gate](https://sonarqube.com/api/badges/gate?key=com.codenotfound:spring-kafka-multiple-topics)](https://sonarqube.com/dashboard/index/com.codenotfound:spring-kafka-multiple-topics)
+
+A detailed step-by-step tutorial on how to send/receive JSON messages using Spring Kafka and Spring Boot.
+
+[https://www.codenotfound.com/spring-kafka-consume-multiple-topics-example.html](https://www.codenotfound.com/spring-kafka-consume-multiple-topics-example.html)

spring-kafka-multiple-topics/README.md

@@ -4,12 +4,12 @@
   <modelVersion>4.0.0</modelVersion>
 
   <groupId>com.codenotfound</groupId>
-  <artifactId>spring-kafka-multiple-consumers</artifactId>
+  <artifactId>spring-kafka-multiple-topics</artifactId>
   <version>0.0.1-SNAPSHOT</version>
 
-  <name>spring-kafka-multiple-consumers</name>
-  <description>Spring Kafka - Multiple Consumers Example</description>
-  <url>https://www.codenotfound.com/spring-kafka</url>
+  <name>spring-kafka-multiple-topics</name>
+  <description>Spring Kafka - Consume Multiple Topics Example</description>
+  <url>https://www.codenotfound.com/spring-kafka-consume-multiple-topics-example.html</url>
 
   <parent>
     <groupId>org.springframework.boot</groupId>

spring-kafka-multiple-consumers/pom.xml renamed to spring-kafka-multiple-topics/pom.xml

@@ -19,13 +19,13 @@ public CountDownLatch getLatch() {
     return latch;
   }
 
-  @KafkaListener(topics = "${topic.bar}")
+  @KafkaListener(topics = "${kafka.topic.bar}")
   public void receiveBar(Bar bar) {
     LOGGER.info("received {}", bar.toString());
     latch.countDown();
   }
 
-  @KafkaListener(topics = "${topic.foo}")
+  @KafkaListener(topics = "${kafka.topic.foo}")
   public void receiveFoo(Foo foo) {
     LOGGER.info("received {}", foo.toString());
     latch.countDown();

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/kafka/SpringKafkaApplication.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/kafka/SpringKafkaApplication.java

@@ -4,7 +4,8 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.messaging.Message;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.support.MessageBuilder;
 
 public class Sender {
 
@@ -13,8 +14,9 @@ public class Sender {
   @Autowired
   private KafkaTemplate<String, ?> kafkaTemplate;
 
-  public void send(Message<?> message) {
-    LOGGER.info("sending {}", message.toString());
-    kafkaTemplate.send(message);
+  public void send(String topic, Object payload) {
+    LOGGER.info("sending payload='{}' to topic='{}'", payload.toString(), topic);
+    kafkaTemplate
+        .send(MessageBuilder.withPayload(payload).setHeader(KafkaHeaders.TOPIC, topic).build());
   }
 }

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/kafka/consumer/Receiver.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/kafka/consumer/Receiver.java

@@ -22,7 +22,6 @@ public class SenderConfig {
   @Bean
   public Map<String, Object> producerConfigs() {
     Map<String, Object> props = new HashMap<>();
-
     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);

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/kafka/consumer/ReceiverConfig.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/kafka/consumer/ReceiverConfig.java

@@ -0,0 +1,5 @@
+kafka:
+  bootstrap-servers: localhost:9092
+  topic:
+    bar: bar.t
+    foo: foo.t

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/kafka/producer/Sender.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/kafka/producer/Sender.java

@@ -10,7 +10,6 @@
   <logger name="com.codenotfound" level="INFO" />
   <logger name="org.springframework" level="WARN" />
 
-
   <root level="WARN">
     <appender-ref ref="STDOUT" />
   </root>

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/kafka/producer/SenderConfig.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/kafka/producer/SenderConfig.java

@@ -5,19 +5,15 @@
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Before;
-import org.junit.BeforeClass;
 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.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.listener.MessageListenerContainer;
-import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import com.codenotfound.kafka.consumer.Receiver;
@@ -44,11 +40,6 @@ public class SpringKafkaApplicationTest {
   @ClassRule
   public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, BAR_TOPIC, FOO_TOPIC);
 
-  @BeforeClass
-  public static void setUpBeforeClass() throws Exception {
-    System.setProperty("kafka.bootstrap-servers", embeddedKafka.getBrokersAsString());
-  }
-
   @Before
   public void setUp() throws Exception {
     // wait until the partitions are assigned
@@ -61,13 +52,8 @@ public void setUp() throws Exception {
 
   @Test
   public void testReceive() throws Exception {
-    Message<Bar> bar =
-        MessageBuilder.withPayload(new Bar("bar")).setHeader(KafkaHeaders.TOPIC, BAR_TOPIC).build();
-    sender.send(bar);
-
-    Message<Foo> foo =
-        MessageBuilder.withPayload(new Foo("foo")).setHeader(KafkaHeaders.TOPIC, FOO_TOPIC).build();
-    sender.send(foo);
+    sender.send(BAR_TOPIC, new Bar("bar"));
+    sender.send(FOO_TOPIC, new Foo("foo"));
 
     receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
     assertThat(receiver.getLatch().getCount()).isEqualTo(0);

spring-kafka-multiple-consumers/src/main/java/com/codenotfound/model/Bar.java renamed to spring-kafka-multiple-topics/src/main/java/com/codenotfound/model/Bar.java

@@ -0,0 +1,5 @@
+kafka:
+  bootstrap-servers: ${spring.embedded.kafka.brokers}
+  topic:
+    bar: bar.t
+    foo: foo.t