Improve kafka integration

starters/kafka-spring-boot-starter/src/main/java/com/schbrain/framework/autoconfigure/kafka/CustomKafkaListenerEndpointRegistry.java

@@ -14,19 +14,20 @@ public class CustomKafkaListenerEndpointRegistry extends KafkaListenerEndpointRe
 
     private static final String CONSUMER_ENABLED_KEY = "spring.kafka.consumer.enabled";
 
-    private final KafkaProperties kafkaProperties;
+    private final boolean shouldRegisterConsumer;
 
     public CustomKafkaListenerEndpointRegistry(KafkaProperties kafkaProperties) {
-        this.kafkaProperties = kafkaProperties;
+        this.shouldRegisterConsumer = EnvUtils.runningOnCloudPlatform() || kafkaProperties.getConsumer().isEnabled();
+        if (!shouldRegisterConsumer) {
+            log.warn("Not running on CloudPlatform or {} is set to false, will not listen to messages from brokers", CONSUMER_ENABLED_KEY);
+            log.warn("If you want force to register with Kafka Brokers, set {} = true", CONSUMER_ENABLED_KEY);
+        }
     }
 
     @Override
     public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory, boolean startImmediately) {
-        if (EnvUtils.runningOnCloudPlatform() || kafkaProperties.getConsumer().isEnabled()) {
+        if (shouldRegisterConsumer) {
             super.registerListenerContainer(endpoint, factory, startImmediately);
-        } else {
-            log.warn("Not running on CloudPlatform or {} is set to false, will not listen to messages from brokers", CONSUMER_ENABLED_KEY);
-            log.warn("If you want force to register with Kafka Brokers, set {} = true", CONSUMER_ENABLED_KEY);
         }
     }
 

starters/starrocks-spring-boot-starter/src/main/java/com/schbrain/framework/autoconfigure/starrocks/helper/ConvertUtils.java

@@ -18,12 +18,20 @@ public class ConvertUtils {
 
     private static final ObjectMapper DESERIALIZER = JacksonUtils.getObjectMapper().copy().setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
 
-    public static <Source, Target> Target convertTo(ConsumerRecord<String, String> record, Class<Source> sourceType, Class<Target> targetType) {
+    public static <Target> Target convertTo(ConsumerRecord<String, String> record, Class<Target> targetType) {
         CanalChangedEvent event = JacksonUtils.getObjectFromJson(record.value(), CanalChangedEvent.class);
         if (event == null) {
             throw new BaseException("CanalChangedEvent is null");
         }
-        Source source = DESERIALIZER.convertValue(event.getAfter(), sourceType);
+        return DESERIALIZER.convertValue(event.getAfter(), targetType);
+    }
+
+    public static <Target> List<Target> convertToList(ConsumerRecords<String, String> records, Class<Target> targetType) {
+        return StreamUtils.toList(records, record -> convertTo(record, targetType));
+    }
+
+    public static <Source, Target> Target convertTo(ConsumerRecord<String, String> record, Class<Source> sourceType, Class<Target> targetType) {
+        Source source = convertTo(record, sourceType);
         return BeanCopyUtils.copy(source, targetType);
     }
 

support/schbrain-spring-support/src/main/java/com/schbrain/framework/support/spring/env/DefaultPropertiesEnvironmentPostProcessor.java

@@ -61,6 +61,8 @@ public class DefaultPropertiesEnvironmentPostProcessor extends LoggerAwareEnviro
         defaultProperties.put("spring.mvc.format.date-time", DatePattern.NORM_DATETIME_PATTERN);
         defaultProperties.put("spring.jackson.date-format", DatePattern.NORM_DATETIME_PATTERN);
         defaultProperties.put("spring.jackson.time-zone", TimeZone.getDefault().getID());
+        // kafka
+        defaultProperties.put("spring.kafka.consumer.group-id", ApplicationName.get(environment));
         // others
         defaultProperties.put("spring.mandatory-file-encoding", StandardCharsets.UTF_8.name());
         defaultProperties.put("spring.web.resources.add-mappings", false);