kafka transaction(exactly once semantic)
kafka transaction 설명에 앞서 kafka 통신을 위한 기본적인 코드부터 설명하려 한다. 테스트 코드에서 EmbeddedKafkaBroker 를 활용해 broker 역할을 대신했다. 그리고 json 데이터를 보내기 위해 objectMapper 를 생성했다. 마지막으로 데이터 전달을 위해 Product 객체를 생성했다.
public class KafkaTest {
static EmbeddedKafkaBroker BROKER;
static final String TOPIC_NAME = "ybs-topic";
ObjectMapper objectMapper = JsonMapper.builder().build();
static {
// broker 갯수 1개, 파티션 갯수 1개
BROKER = new EmbeddedKafkaBroker(1, true, 1)
.kafkaPorts(9092)
.brokerProperty(KafkaConfig.OffsetsTopicReplicationFactorProp(), (short)1)
.brokerProperty("transaction.state.log.replication.factor", (short)1)
.brokerProperty("transaction.state.log.min.isr", 1);
ROKER.afterPropertiesSet();
}
@Getter
@ToString
private static class Product {
private String name;
private int price;
@ConstructorProperties({"name", "price"})
public Product(String name, int price) {
this.name = name;
this.price = price;
}
}
}
다음으로 kafka transaction 개념부터 살펴보자. kafka transaction 은 consuming 한 것에 대해 다시 producing 할 때, 하나로 묶는 역할로써 의미가 있다.
아래 그림에서 Service1 이 produce 한 메세지를 Service2 가 consume 하고, 그 메세지를 갖고 추가 작업을 수행한 Service3 에서 다시 producing 했을 때, producing 이 성공해야만 이전에 consuming 한것도 offset 이 성공했다고 찍힌다(ack가).
다시말해 producing 성공과 consume ack offset 이 transaction 으로 묶여, producing 이 실패하면 offset 도 성공했다고 안찍히기 때문에 다시 consming 을 하게 된다.
cf) Service2 에서 Service3 으로 delegate 하는 것은 같은 애플리케이션 내 메서드 이동을 생각하고 표현했다. 만약 Service3 이 외부 모듈이고 네트워크를 타는 API가 필요하면 실패 했을 때 고려해야 될게 더 많아진다.
이제 kafka producer, consumer property 설정부터 살펴보자. 모든 설정을 언급하진 않고 몇개만 골라서 설명하려 한다.
@Test
void kafkaExactlyOnceSemanticTest() throws JsonProcessingException {
/*
* producer
*/
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER.getBrokersAsString());
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
// producer transaction 설정
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
/*
* consumer
*/
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER.getBrokersAsString());
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
// 같은 그룹의 consumer 들은 offset 공유하므로 같은 메세지를 받지 않음
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "sample-consumer-group");
// latest : consumer group에 새로운 consumer가 붙으면, 새로운 consumer 붙은 순간부터 읽겠다
// earliest : consumer group에 새로운 consumer가 붙으면, topic 에 해당하는 것들을 처음부터 다 읽겠다.
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
// 필수값은 아닌데, 문제 생겼을 때 디버깅할 때 좋음. 다른 컨슈버들간 구분하기 좋음.
consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
// consumer transaction 설정
// default : true
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
// read_committed: 커밋된 데이터만 읽는다.
// default : read_uncommitted
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
}
먼저 ENABLE_AUTO_COMMIT_CONFIG 설정을 좀 자세히 살펴보자. ENABLE_AUTO_COMMIT_CONFIG 이 false 면 consume 을 한 후에 비지니스 처리 하고 commit 을 직접 찍어줘야한다. 만약 true 면 자동으로 offset에 commit을 시킨다. 하지만 위험하다. 만약 작업 중간에 에러가 발생하면 그 때의 offset 부터 다시 읽어야 하는데 자동적으로 커밋을 해버리니까 다음콜을 했을 때 이미 이전꺼는 commit 이 되버렸기 때문에 메세지가 유실될 위험이 있다. 그래서 직접 커밋을 제어하는 방향이 안전하다. 유실보다는 중복이 낫다.
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
다음으로 kafka transaction 을 위해 필요한 설정들을 살펴보자. producer 에서 필요한 설정은 두개다. 첫째, TRANSACTIONAL_ID_CONFIG 는 producer transaction 을 구분하기 위한 id 이고 여기선 UUID 를 사용했다.
// producer transaction 설정
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());
둘째, ENABLE_IDEMPOTENCE_CONFIG 는 true 로 지정한다. produce send 해서 메세지가 kafka broker 에 들어는 갔는데 timeout 이나 broker failure 로 실패응답을 받았다고 해보자. 그러면 producer 는 같은 메세지를 재발행 한다. producer 입장에서는 실패 응답을 받았으므로 메세지가 kafka broker 에 들어갔는지 안들어갔는지 알 수가 없기 때문이다. consumer 입장에서는 중복 발행이 된거기 때문에 똑같은 메세지를 2번 받게 된다. 코드상으로는 한번만 send 했지만 내부 구조상 재발행 하게 되어있다. 여기서 ENABLE_IDEMPOTENCE_CONFIG 가 true 로 되어 있으면, 재발행되서 두번째 메세지가 갔을 때 같은 메세지인지 비교한다. 그리고 두번째로 온 메세지가 이전과 같은 메세지라면 버린다.
// producer transaction 설정
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
consumer 에서 필요한 설정도 두개다. ENABLE_AUTO_COMMIT_CONFIG 는 위에서 설명했다.
ISOLATION_LEVEL_CONFIG 는 read_committed 으로 설정해, producer 가 broker 로 보낸 데이터 중 transaction 이 완벽하게 완료된(커밋된) 데이터에 대해서만 읽을 수 있다. 만약 read_committed 가 아니면, 메세지는 보내졌는데 커밋이 실패했을 때 consumer poll 했을 때 메세지를 읽어오게 된다. 이거는 transaction 을 사용하는 의도와 맞지 않기 때문에 read_committed 를 해줘야 poll 할 때, 이미 나가버린 메세지가 커밋이 안되 있으면 안읽고 skip 시킨다.
// consumer transaction 설정
// default : true
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
// read_committed: 커밋된 데이터만 읽는다.
// default : read_uncommitted
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
이제 kafka transaction 작업한 코드를 보자. 먼저 consumer 객체를 만들어서 while 문 안에서 poll 작업을 수행한다. 그런데 transaction 시나리오 상 consume 하고 다시 produce 해야하므로, consume 이 동작하기 위한 최초의 produce 작업이 필요하다. 그래서 flag 값으로 한번만 수행되는 produce 작업을 만들었다.
그 후로는 produce 된 메세지를 consumer 가 읽어 records.forEach 메서드를 통해 다시 producing 하는 순환구조다. 그리고 initTransactions 메서드로 transaction 처리 준비를 하고 send 전 beginTransaction 메서드, send 후 sendOffsetsToTransaction, commitTransaction 메서드를 수행한다.
cf) 실제로 만들때는 produce 작업과 consume 작업이 분리시킨다. 그래서 한곳에서 같이 쓸일은 없지만 transation 테스트를 위해 만들었다. 위의 그림에서 Service1, Service2, Service3 역할을 한번에 다 하기 때문에 코드가 복잡하다.
try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
consumer.subscribe(Collections.singletonList(TOPIC_NAME));
// cf) consumer.assign() 으로 consume 할 특정 토픽 특정 파티션을 직접 지정할 수 있다.
boolean flag = true;
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
Producer<String, String> producer = new KafkaProducer<>(producerProps);
producer.initTransactions();
producer.beginTransaction();
// 트랜잭션 시나리오는 consumer 부터 시작이므로, 최초 한번 producing 작업이 필요하다.
if (flag) {
producer.send(makeProducerRecord("toy", 1000));
producer.flush();
flag = false;
}
records.forEach(rec -> {
try {
UUID key = this.objectMapper.readValue(rec.key(), UUID.class);
Product value = this.objectMapper.readValue(rec.value(), Product.class);
int partition = rec.partition();
System.out.println("key : " + key);
System.out.println("value : " + value);
System.out.println("partition : " + partition);
producer.send(makeProducerRecord(value.getName(), value.getPrice() + 1));
} catch (IOException ex) {
throw new RuntimeException(ex);
}
});
producer.sendOffsetsToTransaction(
Collections.singletonMap(new TopicPartition(TOPIC_NAME, 0), new OffsetAndMetadata(1)),
new ConsumerGroupMetadata("group")
);
producer.commitTransaction(); // 커밋하면 자동으로 flush 해주게 됌
}
}
private ProducerRecord<String, String> makeProducerRecord(String name, int price) throws JsonProcessingException {
return new ProducerRecord<>(
TOPIC_NAME,
0,
Instant.now().toEpochMilli(),
this.objectMapper.writeValueAsString(UUID.randomUUID()),
this.objectMapper.writeValueAsString(new Product(name, price))
);
}
결과는 아래와 같이 출력되 나온다.
key : f843dfec-1fcc-4be2-a57c-93d54e490c7f
value : Product{name='toy', price=1000}
partition : 0
key : 5a0e4619-8d21-4960-996c-35611a7d5314
value : Product{name='toy', price=1001}
partition : 0
key : 37aa39f9-1eb7-41ea-8c7d-2ff714b9570a
value : Product{name='toy', price=1002}
partition : 0
key : 76d4affd-c0d0-4e40-a55a-13ec86cd0462
value : Product{name='toy', price=1003}
partition : 0
cf ) 추가한 의존성.
dependencies {
implementation("org.springframework.kafka:spring-kafka:2.8.1")
implementation("com.fasterxml.jackson.core:jackson-databind:2.13.1")
testImplementation("org.springframework.kafka:spring-kafka:2.8.1")
testImplementation("org.springframework.kafka:spring-kafka-test:2.8.1")
testImplementation 'org.junit.jupiter:junit-jupiter-api:5.7.0'
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.7.0'
}