프로듀서, 컨슈머 애플리케이션, 그 외 몇가지 궁금한게 있습니다.

안녕하세요. 문의주신 내용 답변드리기 전에 우선, 인프런 강의는 스프링관련 내용은 포함되어 있지 않으므로 관련 내용에 대해 자세히 답변드리기 어려운점 참고부탁드립니다. 대신 일반적인 카프카 클라이언트 사용방식을 배경으로 답변드리도록 하겠습니다.

1.프로듀서 애플리케이션에서 데이터 저장과 레코드 전송을 원자성 단위로 묶는 방법

특정 데이터를 디비에 저장하고 이와 동시에 프로듀서 레코드를 전송하는 방식에 대해 문의하셨는데요. 이미 아시다시피 이종 데이터베이스 간에 원자성은 관련 기능을 제공하지 않는 경우에는 달성하기 어렵습니다. 예를 들어 mysql에 데이터를 저장함과 동시에 프로듀서로 레코드를 전송하는 것을 하나의 트랜잭션으로 묶는 것은 불가능합니다. 그렇기 때문에 특정시점에 장애가 발생할 경우 데이터베이스에는 데이터가 저장되고 프로듀서에는 전송이 안되거나 프로듀서에는 전송이 되었지만 데이터가 저장되지 않은 경우도 발생할 가능성은 있습니다.

그렇기 때문에 이런 상황을 해결하기 위해서는 두가지 해결 방안이 있을것 같습니다.

방법1) 유실이나 중복을 감안하고 개발 : 정확히 한 번 처리하지 않아도 될 때 사용하는 방법입니다. 장애가 발생했을 경우 유실/중복이 발생하더라도 비즈니스적으로 이슈가 없을 경우 적용할 수 있습니다.

방법2) 배치처리를 통해 데이터의 유실/중복을 감지하고 처리하는 로직 개발 : 또 다른 모듈을 사용/개발하여 유실이나 중복이 생기지 않았는지 확인하고 처리하는 방법입니다. 비용은 많이 들지만, 정확히 한 번 처리할 수 있다는 장점이 있습니다.

2.컨슈머 레코드 처리 실패시 커밋 처리 방법

컨슈머의 레코드 처리가 실패가 된다고 해서 계속해서 해당 레코드를 처리 시도를 하게 되면 해당 파티션의 데이터 처리가 늦어지게되어 컨슈머 랙이 상승할 수 있습니다. 이를 방지하기 위해 DLQ(Dead Letter Queue)와 같은 방어로직을 위한 토픽을 따로 만들어 운영하는 경우도 많습니다. 그래서 에러가 발생한 레코드는 DLQ 토픽으로 전송하되 다음 데이터를 처리하고 커밋을 이어나가는 방식이 일반적일것 같습니다.

3.데이터베이스 저장시 컨슈머의 커넥션 유지관리 방법

커넥션풀의 개수가 부담되는 상황이라면 말씀대로 직접 커넥션을 생성하고 해제하는 방식도 괜찮아보입니다. 다만, 성능상에 어떤 이슈가 있을 지는 테스트가 필요해보입니다. 그리고 내부 로직에 따라 1개 컨슈머가 10개 커넥션풀을 모두 사용할 수도 있으므로 판단은 해당 로직 개발자의 의도를 파악해야 할 것 같습니다.

4.카프카를 활용한 보상 트랜잭션 처리 방법

네 가능합니다. 2019년에 열린 런던 카프카 써밋에서 발표한 아래 발표자료와 영상을 보시면 좋을것 같습니다.

• https://www.confluent.io/kafka-summit-lon19/simplifying-distributed-transactions-sagas-kafka/

안녕하세요, 인프런 AI 인턴입니다.

화이팅님의 궁금증에 대해 답변드리겠습니다.

첫 번째로 프로듀서 애플리케이션의 경우, 스프링에서는 KafkaTransactionManager를 사용하여 데이터베이스 트랜잭션과 카프카 레코드 전송을 하나의 트랜잭션으로 묶을 수 있습니다. 먼저 spring.kafka.producer.transaction-id-prefix 프로퍼티를 설정하여 트랜잭션을 활성화시키고, @Transactional 어노테이션을 사용한 메소드 내에서 KafkaTemplate를 이용해 메시지를 전송합니다. 만약 트랜잭션 중 에러가 발생하면 전체 트랜잭션이 롤백되어 데이터베이스 뿐만 아니라 카프카 브로커로의 레코드 전송도 되지 않습니다. 예시 코드는 다음과 같습니다.

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.beans.factory.annotation.Autowired;

public class UserService {

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  @Transactional
  public void save() {
    // 데이터베이스 저장 로직...
    kafkaTemplate.send("someTopic", "Hello");
    // 필요한 경우 다른 에러 핸들링을 수행...
  }
}

더 자세한 사항은 여기의 Spring Boot Reference Documentation을 참고하세요.

두 번째로, 컨슈머 애플리케이션은 @KafkaListener 어노테이션을 통해 메시지를 수신합니다. 처리 중 에러가 발생할 경우, CommonErrorHandler, AfterRollbackProcessor와 같은 스프링 제공 인터페이스를 구현하여 실패한 메시지에 대한 후처리 로직을 구현할 수 있습니다. 예를 들어, 실패한 메시지를 별도의 토픽으로 전송하거나, 로깅 후 다음 메시지로 건너뛰는 등의 전략을 적용할 수 있습니다. 참고로, CommonErrorHandler 인터페이스는 Spring for Apache Kafka version 2.7 이상부터 지원됩니다. 여기에서 더 자세한 정보를 확인할 수 있습니다.

세 번째 질문에 대한 답변입니다. 히카리 커넥션 풀의 크기를 조절하거나, 파티션에 맞게 커넥션 풀을 최적화하는 방법을 활용할 수 있습니다. 대규모 트래픽을 처리하는 애플리케이션의 경우, 커넥션 풀 크기를 조절하거나 다중 인스턴스로 올바르게 분산 처리를 해야합니다. 커넥션 풀 당 커넥션 수를 조정하거나, 여러 컨슈머 인스턴스에서 하나의 커넥션 풀을 공유하도록 설계할 수도 있습니다. 이 경우에는 커넥션 풀 설정에 대한 성능 테스트와 모니터링을 통해 최적의 설정 값을 찾아야 합니다.

마지막으로, MSA(Microservices Architecture)에서 카프카를 통해 사가 패턴(Saga Pattern)을 구현할 수 있습니다. 각 서비스에서 로컬 트랜잭션을 수행한 후 이벤트를 카프카에 발행하고, 다른 서비스에서 이를 구독하여 필요한 보상 트랜잭션을 수행하는 방식으로 구현할 수 있습니다. 사가 패턴은 분산된 마이크로서비스 환경에서 데이터 일관성을 유지하는 데 유용한 방법입니다.

추가적으로 궁금한 사항이나 더 자세한 설명을 원하시면 언제든지 질문해 주세요.