Apache Kafka 입문부터 실전까지

1
2
// 같은 키(orderId)를 가진 메시지는 동일 파티션으로
producer.send(new ProducerRecord<>("order-events", orderId, event));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());

// acks 설정
props.put(ProducerConfig.ACKS_CONFIG, "all"); // 모든 레플리카 확인

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 비동기 전송
producer.send(new ProducerRecord<>("order-events", orderId, orderJson),
    (metadata, exception) -> {
        if (exception != null) {
            log.error("전송 실패: {}", exception.getMessage());
        } else {
            log.info("전송 성공: topic={}, partition={}, offset={}",
                    metadata.topic(), metadata.partition(), metadata.offset());
        }
    });

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 수동 커밋

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("order-events"));

while (true) {
    ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofMillis(1000));

    for (ConsumerRecord<String, String> record : records) {
        processOrder(record.value());
    }

    // 처리 완료 후 오프셋 커밋
    consumer.commitSync();
}

1
2
3
4
5
6
7
8
9
// 레코드 단위 커밋 (가장 정밀)
for (ConsumerRecord<String, String> record : records) {
    processOrder(record.value());

    consumer.commitSync(Map.of(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1)
    ));
}

1
2
3
4
5
6
// 먼저 커밋 → 처리 중 실패하면 메시지 유실
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
consumer.commitSync(); // 먼저 커밋!
for (ConsumerRecord<String, String> record : records) {
    processOrder(record.value()); // 여기서 실패하면 유실
}

1
2
3
4
5
6
7
// 처리 후 커밋 → 커밋 전 실패하면 재처리(중복)
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    processOrder(record.value()); // 먼저 처리!
}
consumer.commitSync(); // 처리 완료 후 커밋
// 여기서 실패하면 → 다음 poll에서 같은 메시지 다시 수신

1
2
3
4
5
6
7
8
9
public void processOrder(OrderEvent event) {
    // 이미 처리한 이벤트인지 확인
    if (processedEventRepository.existsById(event.getEventId())) {
        log.info("이미 처리된 이벤트: {}", event.getEventId());
        return;
    }
    orderService.createOrder(event);
    processedEventRepository.save(new ProcessedEvent(event.getEventId()));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// Producer: 트랜잭션 설정
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-1");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("order-events", key, value));
    producer.send(new ProducerRecord<>("payment-events", key, paymentValue));
    producer.commitTransaction(); // 두 메시지 모두 성공 또는 모두 실패
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

1
2
// Consumer: read_committed로 커밋된 메시지만 소비
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@Service
@RequiredArgsConstructor
public class OrderEventProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void publishOrderCreated(Order order) {
        OrderEvent event = new OrderEvent(
                UUID.randomUUID().toString(),
                order.getId(),
                "ORDER_CREATED",
                order
        );

        CompletableFuture<SendResult<String, OrderEvent>> future =
                kafkaTemplate.send("order-events", order.getId(), event);

        future.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("메시지 전송 실패: {}", ex.getMessage());
            } else {
                log.info("메시지 전송 성공: topic={}, partition={}, offset={}",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
        });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@Service
@RequiredArgsConstructor
public class OrderEventConsumer {

    private final OrderService orderService;

    @KafkaListener(
            topics = "order-events",
            groupId = "order-service",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.info("수신: event={}, partition={}, offset={}",
                event.getType(), partition, offset);

        try {
            switch (event.getType()) {
                case "ORDER_CREATED" -> orderService.handleCreated(event);
                case "ORDER_PAID" -> orderService.handlePaid(event);
                case "ORDER_CANCELLED" -> orderService.handleCancelled(event);
                default -> log.warn("알 수 없는 이벤트: {}", event.getType());
            }
            ack.acknowledge(); // 수동 커밋
        } catch (Exception e) {
            log.error("이벤트 처리 실패: {}", e.getMessage());
            // ack 하지 않음 → 재시도 또는 DLQ로 이동
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Configuration
public class KafkaConfig {

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            kafkaListenerContainerFactory(
                    ConsumerFactory<String, OrderEvent> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3); // 3개 스레드로 병렬 소비
        factory.getContainerProperties()
                .setAckMode(ContainerProperties.AckMode.MANUAL);

        // 에러 핸들러 + DLQ
        factory.setCommonErrorHandler(
                new DefaultErrorHandler(
                        new DeadLetterPublishingRecoverer(kafkaTemplate),
                        new FixedBackOff(1000L, 3) // 1초 간격, 3번 재시도
                ));

        return factory;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@Configuration
public class KafkaConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
        // DLQ로 보내는 Recoverer
        DeadLetterPublishingRecoverer recoverer =
                new DeadLetterPublishingRecoverer(template,
                        (record, ex) -> new TopicPartition(
                                record.topic() + ".DLT", record.partition()));

        // 1초, 2초, 4초 간격으로 3번 재시도 후 DLQ
        ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
        backOff.setMaxElapsedTime(10000L);

        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

        // 특정 예외는 재시도 없이 바로 DLQ로
        handler.addNotRetryableExceptions(
                InvalidMessageException.class,
                DeserializationException.class
        );

        return handler;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@KafkaListener(topics = "order-events.DLT", groupId = "dlq-handler")
public void handleDlq(
        ConsumerRecord<String, OrderEvent> record,
        @Header(KafkaHeaders.DLT_EXCEPTION_MESSAGE) String errorMessage,
        @Header(KafkaHeaders.DLT_ORIGINAL_TOPIC) String originalTopic) {

    log.error("DLQ 수신: key={}, error={}, originalTopic={}",
            record.key(), errorMessage, originalTopic);

    // 알림 발송 (Slack, PagerDuty 등)
    alertService.sendDlqAlert(record, errorMessage);

    // 필요 시 수동 재처리 큐에 저장
    dlqRepository.save(new DlqRecord(record, errorMessage, originalTopic));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@RetryableTopic(
        attempts = "4",
        backoff = @Backoff(delay = 60000, multiplier = 10, maxDelay = 600000),
        dltStrategy = DltStrategy.FAIL_ON_ERROR,
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE
)
@KafkaListener(topics = "order-events", groupId = "order-service")
public void handleOrderEvent(OrderEvent event, Acknowledgment ack) {
    orderService.process(event);
    ack.acknowledge();
}

@DltHandler
public void handleDlt(OrderEvent event) {
    log.error("최종 실패 — DLT 도착: {}", event);
    alertService.sendCriticalAlert(event);
}

1
2
3
4
5
6
7
8
9
10
11
// Kafka를 이벤트 저장소로 활용
@Service
public class OrderEventStore {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void append(String orderId, OrderEvent event) {
        // 주문 ID를 키로 → 같은 파티션에 순서대로 저장
        kafkaTemplate.send("order-event-store", orderId, event);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
// Command: 주문 생성
@Service
public class OrderCommandService {
    public void createOrder(CreateOrderCommand command) {
        Order order = Order.create(command);
        orderRepository.save(order);

        // 이벤트 발행
        kafkaTemplate.send("order-events", order.getId(),
                new OrderCreatedEvent(order));
    }
}

// Query: 읽기 모델 업데이트
@Service
public class OrderQueryProjector {

    @KafkaListener(topics = "order-events", groupId = "order-query")
    public void project(OrderEvent event) {
        switch (event) {
            case OrderCreatedEvent e -> {
                OrderView view = new OrderView(
                        e.getOrderId(), e.getCustomerName(),
                        e.getItems(), e.getTotal(), "CREATED"
                );
                orderViewRepository.save(view); // Elasticsearch, MongoDB 등
            }
            case OrderShippedEvent e -> {
                orderViewRepository.updateStatus(
                        e.getOrderId(), "SHIPPED", e.getTrackingNo());
            }
            // ...
        }
    }
}

// Query: 읽기 전용 API
@RestController
public class OrderQueryController {

    @GetMapping("/orders/{orderId}")
    public OrderView getOrder(@PathVariable String orderId) {
        return orderViewRepository.findById(orderId).orElseThrow();
    }

    @GetMapping("/orders/search")
    public List<OrderView> searchOrders(@RequestParam String keyword) {
        return orderViewRepository.searchByKeyword(keyword);
    }
}