Methods for Springboot project to consume Kafka data

<dependencies>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId></groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- Spring Kafka -->
    <dependency>
        <groupId></groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Spring Boot Starter for Logging (optional but useful for debugging) -->
    <dependency>
        <groupId></groupId>
        <artifactId>spring-boot-starter-logging</artifactId>
    </dependency>
    <!-- Spring Boot Starter for Testing -->
    <dependency>
        <groupId></groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

spring:
  kafka:
    bootstrap-servers: localhost:9092  # Kafka server address    consumer:
      group-id: my-consumer-group   # Consumer group ID      auto-offset-reset: earliest   # The consumer reads from scratch (if there is no submitted offset)      key-deserializer:   # Set the deserializer of the key      value-deserializer:   # Set the deserializer of value to a string    listener:
      missing-topics-fatal: false    # If the theme does not exist，No fatal errors thrown

-servers=localhost:9092
-id=my-consumer-group
-offset-reset=earliest
-topics-fatal=false
-deserializer=  # Set the deserializer of the key-deserializer=  # set upvalueThe deserializer is a string

spring:
  kafka:
    producer:
      key-serializer:   # Use StringSerializer for Key      value-serializer:   # Use ErrorHandlingSerializer for Value      properties:
        :   # Default JSON Deserialization target type is Order

import ;
import ;
import ;
@Service
@EnableKafka  // Enable Kafka Consumerpublic class KafkaConsumer {
    private final ObjectMapper objectMapper = new ObjectMapper();
    // Listen to the order-topic topic in Kafka    @KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
    public void consumeOrder(String message) {
        try {
            // Deserialize the JSON string into an Order object            Order order = (message, );
            ("Received order: " + order);
        } catch (Exception e) {
            ();
        }
    }
}

import ;
import ;
import ;
@Service
public class KafkaConsumer {
    // Listen to the order-topic topic in Kafka    @KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
    public void consumeOrder(ConsumerRecord<String, String> record) {
        // Get detailed information about the message        String key = ();           // Get the key of the message        String value = ();       // Get the value of the message        String topic = ();       // Get the topic of the message        int partition = (); // Get the partition of the message        long offset = ();      // Get the offset of the message        long timestamp = (); // Get the timestamp of the message        // Process the message (here we just print the message)        ("Consumed record: ");
        ("Key: " + key);
        ("Value: " + value);
        ("Topic: " + topic);
        ("Partition: " + partition);
        ("Offset: " + offset);
        ("Timestamp: " + timestamp);
    }
}

import ;
import ;
import ;
@Service
public class KafkaConsumer {
    // Listen to the order-topic topic in Kafka    @KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
    public void consumeOrder(ConsumerRecord<String, Order> record) {
        // Get detailed information about the message        String key = ();           // Get the key of the message        Order value = ();       // Get the value of the message        String topic = ();       // Get the topic of the message        int partition = (); // Get the partition of the message        long offset = ();      // Get the offset of the message        long timestamp = (); // Get the timestamp of the message        // Process the message (here we just print the message)        ("Consumed record: ");
        ("Key: " + key);
        ("Value: " + value);
        ("Topic: " + topic);
        ("Partition: " + partition);
        ("Offset: " + offset);
        ("Timestamp: " + timestamp);
    }
}

import ;
import ;
@SpringBootApplication
public class KafkaConsumerApplication {
    public static void main(String[] args) {
        (, args);
    }
}

import ;
import ;
import ;
import ;
import ;
@Service
@EnableKafka
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;  // Send a String type message    private ObjectMapper objectMapper = new ObjectMapper();  // Jackson ObjectMapper is used for serialization    // Send order to Kafka    public void sendOrder(String topic, Order order) {
        try {
            // Convert the Order object to a JSON string            String orderJson = (order);
            // Send JSON string to Kafka            (topic, orderJson);  // Send string message            ("Order JSON sent to Kafka: " + orderJson);
        } catch (Exception e) {
            ();
        }
    }
}

import ;
import ;
import ;
import ;
@Service
@EnableKafka
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Order> kafkaTemplate;
    // Send order to Kafka    public void sendOrder(String topic, Order order) {
        (topic, order);  // Send an order object, Spring Kafka will automatically convert Order to JSON    }
}

bin/ config/

import ;
import ;
import ;
@RestController
public class KafkaController {
    @Autowired
    private KafkaProducer kafkaProducer;
    @GetMapping("/sendOrder")
    public String sendOrder() {
        Order order = new Order();
        (1L);
        (123L);
        ("Laptop");
        (2);
        ("Created");
        ("order-topic", order);
        return "Order sent!";
    }
}