메시지는 보냈는데 DB는 롤백됐다면? — 아웃박스 패턴과 재시도 토픽으로 만드는 자가 치유 파이프라인

api("io.namastack:namastack-outbox-starter-jpa:0.3.0")

outbox:
  poll-interval: 2000
  batch-size: 10
  processing:
    stop-on-first-failure: true
    publish-after-save: true
    delete-completed-records: false
    executor-core-pool-size: 4
    executor-max-pool-size: 8
  retry:
    max-retries: 3
    policy: "exponential"
    exponential:
      initial-delay: 2000
      max-delay: 60000
      multiplier: 2.0

@Override
@Transactional
public void process(Request request) {
    inboundExpectationInterfaceMasterPort.save(request.toMaster());
    inboundExpectationInterfaceDetailPort.saveAll(request.toDetail());
 
    InboundExpectationMessage message = InboundExpectationMessage.from(request);
 
    try {
        OutboxRecord outboxRecord = new OutboxRecord.Builder()
            .aggregateId(message.inboundId())
            .eventType("INBOUND_EXPECTATION_SAVE")
            .payload(objectMapper.writeValueAsString(message))
            .build(clock);
        outboxRecordRepository.save(outboxRecord);
    } catch (Exception e) {
        throw new InboundServiceException(
            "Failed to serialize Inbound Expectation Created Event", e);
    }
}

@Component
@Slf4j
public class KafkaProduceOutboxProcessor implements OutboxRecordProcessor {
    private final ObjectMapper objectMapper;
    private final KafkaProducePort kafkaProducePort;
 
    @Override
    public void process(@NonNull OutboxRecord outboxRecord) {
        switch (outboxRecord.getEventType()) {
            case "INBOUND_EXPECTATION_SAVE":
                this.sendInboundExpectationMessage(outboxRecord);
                break;
            default:
                log.warn("Unsupported OutboxRecord event type: {}",
                    outboxRecord.getEventType());
        }
    }
 
    private void sendInboundExpectationMessage(OutboxRecord outboxRecord) {
        try {
            InboundExpectationMessage message = objectMapper.readValue(
                outboxRecord.getPayload(), InboundExpectationMessage.class);
            kafkaProducePort.send(message);
        } catch (Exception e) {
            log.error("Failed to process OutboxRecord. outboxId: {}, error: {}",
                outboxRecord.getId(), e.getMessage(), e);
        }
    }
}

@RetryableTopic(
    attempts = "145",
    backoff = @Backoff(delay = 600000L),
    sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC,
    kafkaTemplate = "retryKafkaTemplate"
)
@KafkaListener(
    topics = "${spring.kafka.topics.main-topic-name}",
    containerFactory = "containerFactory"
)
public void onMessage(SomethingRequestDTO somethingRequestDTO,
                      Acknowledgment acknowledgment,
                      @Header(KafkaHeaders.RECEIVED_TOPIC) String currentTopicName) {
    TopicNameSet topicNameSet = TopicNameSet.of(mainTopicName, currentTopicName);
    somethingConvertUseCase.convert(somethingRequestDTO.toDomain(), topicNameSet);
    acknowledgment.acknowledge();
}
 
@DltHandler
public void handleDeadLetter(SomethingRequestDTO somethingRequestDTO,
                             @Header(KafkaHeaders.RECEIVED_TOPIC) String dltTopicName) {
    alarmUtils.alarm(
        "[외부 채널 입고 정보 -> 재처리 최종 실패. 데드레터 알림]",
        "id: " + somethingRequestDTO.id() + "\nDeadLetter Topic Name: " + dltTopicName
    );
}