chore(demo-kafka): fix org.apache.kafka.common.errors.TimeoutException

Author: helltractor

demo-kafka/pom.xml

@@ -3,6 +3,7 @@
          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>
+
     <parent>
         <groupId>com.helltractor.demo</groupId>
         <artifactId>parent</artifactId>

demo-kafka/src/main/java/com/helltractor/demo/KafkaApplication.java

@@ -5,7 +5,7 @@
 
 @SpringBootApplication
 public class KafkaApplication {
-    
+
     public static void main(String[] args) {
         SpringApplication.run(KafkaApplication.class, args);
     }

demo-kafka/src/main/java/com/helltractor/demo/message/AbstractMessage.java

@@ -6,12 +6,12 @@
  * Base message object for extends.
  */
 public class AbstractMessage implements Serializable {
-    
+
     /**
      * Reference id, or null if not set.
      */
     public String refId = null;
-    
+
     /**
      * Message create time
      */

demo-kafka/src/main/java/com/helltractor/demo/messaging/BatchMessageHandler.java

@@ -1,11 +1,11 @@
 package com.helltractor.demo.messaging;
 
-import com.helltractor.demo.message.AbstractMessage;
-
 import java.util.List;
 
+import com.helltractor.demo.message.AbstractMessage;
+
 @FunctionalInterface
 public interface BatchMessageHandler<T extends AbstractMessage> {
-    
+
     void processMessages(List<T> messages);
 }

demo-kafka/src/main/java/com/helltractor/demo/messaging/MessageConsumer.java

@@ -2,6 +2,6 @@
 
 @FunctionalInterface
 public interface MessageConsumer {
-    
+
     void stop();
 }

demo-kafka/src/main/java/com/helltractor/demo/messaging/MessageProducer.java

@@ -1,17 +1,17 @@
 package com.helltractor.demo.messaging;
 
-import com.helltractor.demo.message.AbstractMessage;
-
 import java.util.List;
 
+import com.helltractor.demo.message.AbstractMessage;
+
 @FunctionalInterface
 public interface MessageProducer<T extends AbstractMessage> {
-    
+
     default void sendMessages(List<T> messages) {
         for (T message : messages) {
             sendMessages(message);
         }
     }
-    
+
     void sendMessages(T message);
 }

demo-kafka/src/main/java/com/helltractor/demo/messaging/MessageTypes.java

@@ -24,15 +24,15 @@
  */
 @Component
 public class MessageTypes {
-    
+
     private static final char SEP = '#';
-    
+
     private final Logger logger = LoggerFactory.getLogger(getClass());
-    
+
     private final String messagePackage = AbstractMessage.class.getPackageName();
-    
+
     private final Map<String, Class<? extends AbstractMessage>> messageTypes = new HashMap<>();
-    
+
     /**
      * Find all message classes in the package and instantiate them.
      */
@@ -67,13 +67,13 @@ public boolean match(MetadataReader metadataReader, MetadataReaderFactory metada
             }
         }
     }
-    
+
     public String serialize(AbstractMessage message) {
         String type = message.getClass().getName();
         String json = JsonUtil.writeJson(message);
         return type + SEP + json;
     }
-    
+
     public AbstractMessage deserialize(String data) {
         int pos = data.indexOf(SEP);
         if (pos == -1) {
@@ -87,15 +87,15 @@ public AbstractMessage deserialize(String data) {
         String json = data.substring(pos + 1);
         return JsonUtil.readJson(json, clazz);
     }
-    
+
     public List<AbstractMessage> deserialize(List<String> dataList) {
         List<AbstractMessage> list = new ArrayList<>(dataList.size());
         for (String data : dataList) {
             list.add(deserialize(data));
         }
         return list;
     }
-    
+
     public List<AbstractMessage> deserializeConsumerRecords(List<ConsumerRecord<String, String>> dataList) {
         List<AbstractMessage> list = new ArrayList<>(dataList.size());
         for (ConsumerRecord<String, String> data : dataList) {

demo-kafka/src/main/java/com/helltractor/demo/messaging/Messaging.java

@@ -1,27 +1,25 @@
 package com.helltractor.demo.messaging;
 
 public interface Messaging {
-    
+
     enum Topic {
-        
+
         TOPIC_ONE(1),
-        
         TOPIC_TWO(2),
-        
         TOPIC_THREE(3);
-        
+
         final int concurrency;
-        
+
         Topic(int concurrency) {
             this.concurrency = concurrency;
         }
-        
+
         public int getConcurrency() {
             return this.concurrency;
         }
-        
+
         public int getPartitions() {
             return this.concurrency;
         }
     }
-}
+}

demo-kafka/src/main/java/com/helltractor/demo/messaging/MessagingConfiguration.java

@@ -1,5 +1,8 @@
 package com.helltractor.demo.messaging;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -13,25 +16,27 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.core.*;
-
-import java.util.HashMap;
-import java.util.Map;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
 
 @EnableKafka
 @Configuration
 public class MessagingConfiguration {
-    
+
     private final Logger logger = LoggerFactory.getLogger(getClass());
-    
+
     private Map<String, Object> producerConfigs(String bootstrapServers) {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         return configs;
     }
-    
+
     private Map<String, Object> consumerConfigs(String bootstrapServers, int batchSize) {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, Integer.valueOf(batchSize));
@@ -41,27 +46,27 @@ private Map<String, Object> consumerConfigs(String bootstrapServers, int batchSi
         configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         return configs;
     }
-    
+
     @Bean
     public KafkaAdmin admin(@Value("${spring.kafka.bootstrap-servers}") String bootstrapServers) {
         Map<String, Object> configs = new HashMap<>();
         configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         return new KafkaAdmin(configs);
     }
-    
+
     @Bean
     public ProducerFactory<String, String> producerFactory(
             @Value("${spring.kafka.bootstrap-servers}") String bootstrapServers) {
         logger.info("init kafka producer from bootstrap servers: {}", bootstrapServers);
         return new DefaultKafkaProducerFactory<>(producerConfigs(bootstrapServers));
     }
-    
+
     @Bean
     public KafkaTemplate<String, String> kafkaTemplate(@Autowired ProducerFactory<String, String> producerFactory) {
         logger.info("init kafka template...");
         return new KafkaTemplate<>(producerFactory);
     }
-    
+
     @Bean
     public ConsumerFactory<String, String> consumerFactory(
             @Value("${spring.kafka.bootstrap-servers}") String bootstrapServers,
@@ -70,7 +75,7 @@ public ConsumerFactory<String, String> consumerFactory(
                 Integer.valueOf(batchSize));
         return new DefaultKafkaConsumerFactory<>(consumerConfigs(bootstrapServers, batchSize));
     }
-    
+
     @Bean
     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
             @Autowired ConsumerFactory<String, String> consumerFactory) {
@@ -81,4 +86,4 @@ public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerCont
         factory.setBatchListener(Boolean.TRUE);
         return factory;
     }
-}
+}