docs: 📝 [Chapter 07] KafkaProducer 수정정

_posts/kafka/2025-01-04-chapter07.md

@@ -183,8 +183,11 @@ public class KafkaConsumer {
 **`KafkaProducer`**
 
 ```java
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
 import java.util.concurrent.CompletableFuture;
@@ -198,23 +201,51 @@ public class KafkaProducer {
     @Autowired
     public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
         this.kafkaTemplate = kafkaTemplate;
+
+        // ProducerListener 설정
+        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
+
+            @Override
+            public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
+                // 전송 성공 시, RecordMetadata를 통해 전송된 데이터의 메타 정보를 확인
+                String topic = recordMetadata.topic();
+                int partition = recordMetadata.partition();
+                long offset = recordMetadata.offset();
+
+                // bootstrap.servers 출력
+                String bootstrapServers = kafkaTemplate.getProducerFactory().getConfigurationProperties().get("bootstrap.servers").toString();
+                System.out.println("Message successfully sent to topic: " + topic + ", partition: " + partition + ", offset: " + offset);
+            }
+
+            @Override
+            public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
+                // 전송 실패 시, 예외 메시지 출력
+                System.out.println("Failed to send message to topic: " + producerRecord.topic() + ". Exception: " + exception.getMessage());
+                System.out.println("Failed message: " + producerRecord.value()); // 실패한 메시지 출력
+            }
+        });
     }
 
     public void sendMessage(String message) {
+        long startTime = System.currentTimeMillis();
         CompletableFuture<SendResult<String, String>> future = CompletableFuture.supplyAsync(() -> {
             try {
-                return kafkaTemplate.send("my_topic", "key", message).get(); // blocking
+                // 메시지 전송
+                return kafkaTemplate.send("my_topic", "key", message).get();  // blocking
             } catch (InterruptedException | ExecutionException e) {
+                // 실패 시 예외 처리
                 throw new RuntimeException("Failed to send message", e);
             }
         });
 
-        // 비동기 결과 처리
         future.whenComplete((result, ex) -> {
+            long endTime = System.currentTimeMillis();
             if (ex == null) {
-                System.out.println("Message sent successfully: " + result.getProducerRecord().value());
+                System.out.println("Message sent successfully in " + (endTime - startTime) + " ms");
             } else {
-                System.out.println("Failed to send message: " + ex.getMessage());
+                // 지연이나 연결 실패 시 예외 처리
+                System.out.println("Failed to send message in " + (endTime - startTime) + " ms: " + ex.getMessage());
+                // 재시도 로직 추가
             }
         });
     }