@KafkaListener(topicPattern = "TEST.*", groupId = "test")public void consume(GenericRecord record, Acknowledgment ack) { // your logic}

// Add @timestamp field Object originalTimestamp = jsonMap.remove("timestamp"); if (originalTimestamp != null) { jsonMap.put("@timestamp", originalTimestamp); }

topic 생성# broker 서버 접속# topic 목록 확인kafka-topics --bootstrap-server localhost:9092 --list# topic 생성kafka-topics --bootstrap-server localhost:9092 --create --topic test --partitions 1# topic 확인kafka-topics --bootstrap-server localhost:9092 --describe --topic testproducer/consumer 테스트# producer 생성kafka-console-producer --broker-list localhost:9092 --topic test> 여기에 전달할 내용 입력# consumer 생성kafka-conso..