{"time":1745411561314,"blocks":[{"id":"asbW1Qn2TZ","type":"customHeader","data":{"text":"Topic 과 Partition 관계"}},{"id":"BbLX12gMGC","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":" 토픽(Topic): 메시지를 구분하는 카테고리나 주제입니다. 토픽은 연관된 이벤트를 묶어 영속화 하는데, 각 토픽은 하나 이상의 파티션을 가질 수 있습니다. ","meta":{},"items":[]}]}},{"id":"QYlAeccFe6","type":"image","data":{"caption":"","withBorder":true,"withBackground":false,"stretched":false,"file":{"url":"https://backend.owl-dev.me/images/topic_and_partition.png"}}},{"id":"M6Dr2gw52w","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"Kafka 의 구조상 토픽과 파티션은 1:N 관계로 설계되어 있습니다. ","meta":{},"items":[]}]}},{"id":"jzZMkY8yiB","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"파티션 (Partition): 토픽 내에서 메시지를 분산 저장하는 단위입니다. 파티션은 데이터를 분산하고 병렬처리를 가능하게 하는 중요한 역할을 합니다. 각 파티션은 로그 형태로 메시지를 저장하며, 각 메시지는 파티션 내에서 유일한 오프셋 (offset) 을 가집니다. ","meta":{},"items":[]}]}},{"id":"W24iJF-HIF","type":"image","data":{"caption":"","withBorder":true,"withBackground":false,"stretched":true,"file":{"url":"https://backend.owl-dev.me/images/a_view_of_kafka_topic_partition.png"}}},{"id":"bEuPZpdTNo","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"파티션의 레코드는 각각이 Offset 이라 불리며, 각 파티션 내에서의 순서는 보장되지만, 파티션 간에는 순서가 보장되지 않습니다. 그리고 동일한 Topic 을 구독하는 파티션이 3개라면 각 파티션은 보통 다른 브로커에 분할되어서 관리됩니다.","meta":{},"items":[]},{"content":"한 토픽에 대해서 kafka 브로커의 수보다 더 많은 파티션을 가질 수 있으며, Kafka는 토픽의 파티션을 여러 브로커에 걸쳐 분산시켜 저장할 수 있습니다.","meta":{},"items":[]}]}},{"id":"_g3NCsxPnn","type":"customHeader","data":{"text":"Partition 과 Consumer Group 관계"}},{"id":"tNvWBEacb6","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"컨슈머 그룹 (Consumer Group): 파티션은 하나 혹은 그 이상의 Consumer 들로부터 소비될 수 있고, 각각은 서로 다른 Offset 을 가지고 메시지를 읽을 수 있습니다. 관련해서 Consumer Group 이라는 개념을 사용하는데 어떤 토픽을 소비하는 Consumer 들을 그룹으로 묶은 것 입니다.","meta":{},"items":[]},{"content":"동일한 Consumer Group 에 있는 Consumer 들은 동일한 Group-id 를 부여받으며, 파티션 내에 메시지는 그룹 내의 Consumer 중 하나에 의해서만 소비되는 것을 보장합니다.","meta":{},"items":[]}]}},{"id":"IVY-SpZi9O","type":"image","data":{"caption":"","withBorder":true,"withBackground":false,"stretched":true,"file":{"url":"https://backend.owl-dev.me/images/Consumer_Group.png"}}},{"id":"1F0G0XPtVo","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"Consumer Group 은 Consumer 들이 병렬적으로 메시지를 소비할 수 있도록 합니다. 병렬처리를 함으로써 처리량은 높아지지만, 그룹의 최대 병렬 개수는 토픽의 파티션 개수와 동일합니다.","meta":{},"items":[]}]}},{"id":"nwygnm1TMh","type":"image","data":{"caption":"","withBorder":true,"withBackground":false,"stretched":true,"file":{"url":"https://backend.owl-dev.me/images/Consumer_Group2.png"}}},{"id":"svYfiLfYhz","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"예를 들어, N 개의 파티션이 존재하는 토픽에 N+1 개의 Consumer 가 있는 경우, N 개의 Consumer 들은 파티션을 각각 배정받지만, 남은 Consumer 가 장애가 발생하지 않는 한, IDLE 상태로 파티션에 배정되기를 기다리게 됩니다. 이는 Hot Failover 를 구현하는 좋은 전략이 될 수 있습니다.","meta":{},"items":[]}]}},{"id":"asWbuBYEz7","type":"customHeader","data":{"text":"Consumer Group 과 Offset 관계"}},{"id":"CqbcxMOULj","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"오프셋 (Offset): 오프셋은 컨슈머가 특정 파티션에서 마지막으로 읽은 메시지의 위치를 나타내는 값 입니다. 컨슈머 그룹 내의 컨슈머가 메시지를 읽을 때, 그 오프셋을 kafka 에 저장하며, 다음에 메시지를 읽을 때 어디에서부터 시작해야 하는지 알 수 있습니다.","meta":{},"items":[]},{"content":"이 오프셋 정보는 '__consumer__offsets' 라는 시스템 토픽에 저장됩니다. 각 컨슈머 그룹에 대한 오프셋은 이 시스템 토픽 내의 특정 파티션에 저장되며, 컨슈머 그룹이 메시지 처리 상태를 관리할 수 있게 합니다.","meta":{},"items":[]}]}},{"id":"4yhnZIdWZL","type":"image","data":{"caption":"","withBorder":false,"withBackground":false,"stretched":false,"file":{"url":"https://backend.owl-dev.me/images/kafka_offset.png"}}},{"id":"VIist3ol0j","type":"list","data":{"style":"unordered","meta":{},"items":[{"content":"각 파티션은 특정 토픽에 속하며, 오프셋 관리는 '__consumer_offsets' 토픽을 통해 효율적으로 이루어집니다. kafka 를 사용할 때는 이러한 아키텍처를 이해하고 적절히 활용하는 것이 중요합니다.","meta":{},"items":[]}]}},{"id":"FbFZ5M_U9Y","type":"paragraph","data":{"text":"Apache Kafak 에서 메시지의 중복을 방지하는 여러 방법에 대해서 설명하기 전에, 필요한 개념인 토픽과 파티션의 대략적인 개념에 대해서 설명을 했고, 이제 구체적인 이벤트 중복 방지 방법에 대해서 얘기해보려 합니다."}},{"id":"1ePiGTas0E","type":"delimiter","data":{}},{"id":"z0-3SjUvBu","type":"customHeader","data":{"text":"EOS (Exactly-Once Semantics) 를 통한 중복 방지"}},{"id":"69jftOn4IF","type":"paragraph","data":{"text":"Kafka 0.11 버전 이상에서는 Producer 와 Consumer 모두에서 EOS 를 지원합니다."}},{"id":"6uk0odkVDD","type":"paragraph","data":{"text":"Producer 의 'transactional.id' , 'enable.idempotence' 설정 그리고 Consumer 의 'isolation.level' 의 설정을 통해 구현할 수 있습니다."}},{"id":"3-jYaGj8jQ","type":"header","data":{"text":"Producer 에서 EOS 활성화하기 ","level":3}},{"id":"cvVpZZaWr6","type":"paragraph","data":{"text":"kafka Producer 가 멱등성을 보장하도록 설정하면, 네트워크 등의 문제로 인해 메시지가 중복 전송되는 경우에도 Kafka가 중복을 관리하여 동일한 메시지가 여러번 저장되는 것을 방지할 수 있습니다."}},{"id":"Sr8FzSGT5K","type":"paragraph","data":{"text":"'producer' 설정에서 'enable.idempotence' 를 'true' 로 설정합니다."}},{"id":"2VHsSCVLWi","type":"code","data":{"code":"public class KafkaProducerConfig {\n\n // 프로듀서 설정을 반환하는 메서드\n private Properties producerProperties(String transactionalId) {\n Properties props = new Properties();\n props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, \"localhost:9092\");\n props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());\n props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, \"true\");\n props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);\n // 필요한 추가 설정\n props.put(ProducerConfig.ACKS_CONFIG, \"all\");\n props.put(ProducerConfig.RETRIES_CONFIG, 10);\n props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);\n return props;\n }\n\n public void setupProducers() {\n // 프로듀서1 설정\n Properties producer1Props = producerProperties(\"transactional-id-producer-1\");\n KafkaProducer producer1 = new KafkaProducer<>(producer1Props);\n producer1.initTransactions();\n\n try {\n // 트랜잭션 시작\n producer1.beginTransaction();\n // 메시지 전송 \n producer1.send(new ProducerRecord<>(\"topic\", \"message\"));\n // 트랜잭션 커밋\n producer1.commitTransaction();\n } catch (Exception e) {\n // 문제가 발생한 경우 트랜잭션 중단\n producer1.abortTransaction();\n }\n }\n}\n"}},{"id":"8c2qZq5_SZ","type":"header","data":{"text":"Consumer 에서 EOS 활성화하기","level":3}},{"id":"1m1FgUw9jY","type":"paragraph","data":{"text":"컨슈머에서 EOS 를 활성화하기 위해서는 'isolation.level' 을 'read_commited' 로 설정해야 합니다. 이 설정은 컨슈머가 커밋된 트랜잭션의 메시지만 읽도록 보장합니다."}},{"id":"39GGnetsOT","type":"paragraph","data":{"text":"따라서 프로듀서에서 아직 커밋되지 않은 트랜잭션의 메시지는 컨슈머가 읽지 않게 됩니다."}},{"id":"mhk1UjqKcm","type":"code","data":{"code":"Properties props = new Properties();\nprops.put(\"bootstrap.servers\", \"서버 주소\");\nprops.put(\"group.id\", \"그룹 ID\");\nprops.put(\"key.deserializer\", \"org.apache.kafka.common.serialization.StringDeserializer\");\nprops.put(\"value.deserializer\", \"org.apache.kafka.common.serialization.StringDeserializer\");\nprops.put(\"isolation.level\", \"read_committed\");\n\nKafkaConsumer consumer = new KafkaConsumer<>(props);\nconsumer.subscribe(Arrays.asList(\"토픽\"));"}},{"id":"AIa6jT_AaC","type":"paragraph","data":{"text":"EOS 를 사용하면 kafka 시스템 전반에 걸쳐 데이터의 정확성과 일관성을 높일 수 있으며, 분산 시스템에서 발생할 수 있는 중복 처리나 메시지 손실 문제를 효과적으로 해결할 수 있습니다."}},{"id":"OTc9yVCkHM","type":"header","data":{"text":"transactional.id 주의사항","level":2}},{"id":"lXdeCSK4Nw","type":"paragraph","data":{"text":"위에 Producer 에서 props 로 EOS 를 활성화하기 위해서 transactional.id 값을 부여하도록 설정되어 있는데, 이 값을 설정할 때는 몇가지 중요한 사항을 고려해야 합니다."}},{"id":"droWSgVkFh","type":"header","data":{"text":"유일성","level":3}},{"id":"5ELEhfq4LZ","type":"paragraph","data":{"text":"각 트랜잭션을 유일하게 식별하기 위해 'transactional.id' 는 Kafka 클러스터 내에서 고유해야 합니다. 이는 Kafka 가 해당 ID 를 사용하여 트랜잭션의 상태를 관리하기 때문입니다. 따라서 같은 'transactional.id' 를 가진 여러 프로듀서가 동시에 활동하는 경우 트랜잭션 관리에 혼란이 발생할 수 있습니다."}},{"id":"EEKgpiqnKy","type":"header","data":{"text":"지속성","level":3}},{"id":"T_OSJmyGMj","type":"paragraph","data":{"text":"트랜잭션을 안정적으로 관리하기 위해 'transactional.id' 는 프로듀서 재시작 간에도 지속적으로 사용되어야 합니다. 즉 프로듀서가 재시작 되더라도 동일한 'transactional.id' 를 사용하여 이전 트랜잭션의 상태를 이어갈 수 있습니다. 이는 트랜잭션이 중단되었을때 안정적으로 복구할 수 있게 해줍니다."}},{"id":"m9Y0OBklUt","type":"header","data":{"text":"관련성 ","level":3}},{"id":"Bmi_TPWw77","type":"paragraph","data":{"text":" 'transactional.id' 는 트랜잭션 범위와 관련된 작업을 명확히 식별할 수 있도록 관련성 있는 이름을 사용하는 것이 좋습니다. 예를 들어, 특정 작업 또는 데이터 스트림에 대해 고유한 이름을 사용함으로써, 로그나 모니터링 시에 트랜잭션을 쉽게 식별하고 추적할 수 있습니다."}},{"id":"9Ihylj3YcP","type":"header","data":{"text":"transactional.id 를 따로 관리해야 할까?","level":2}},{"id":"QAcHhVORed","type":"paragraph","data":{"text":"transactional.id 는 Kafka 내부에서 트랜잭션을 관리하기 위해 사용되는 설정 값입니다."}},{"id":"7-NVxElAP8","type":"paragraph","data":{"text":"Producer 가 트랜잭션을 시작, 커밋, 롤백하는 과정에서, 'transactional.id' 를 사용하여 트랜잭션의 일관성을 보장하고 Kafka 클러스터 내의 트랜잭션 코디네이터가 이를 추적합니다."}},{"id":"liGl-7DQ19","type":"paragraph","data":{"text":"Producer 나 Consumer 가 장애로 인해 복구 과정을 거칠 때, 사용자가 'transactional.id' 를 직접 관리할 필요는 없습니다. 프로듀서가 재시작되면 설정된 'transactional.id' 를 바탕으로 Kafka 가 자동으로 이전에 시작된 트랜잭션의 상태를 확인하고, 필요한 경우 트랜잭션을 자동으로 재시작하거나 롤백합니다. 이 과정은 모두 Kafka 클러스터 내부적으로 처리되므로, 개발자는 이에 대해 추가적인 처리 로직을 구현할 필요가 없습니다."}},{"id":"OViEAKNsYo","type":"header","data":{"text":"kafka Broker 장애시 EOS 의 한계 ","level":2}},{"id":"gyz88s74rn","type":"paragraph","data":{"text":"EOS 기능은 프로듀서가 메시지를 전송하고 컨슈머가 메시지를 소비하는 과정에서의 중복 처리를 방지하는 메커니즘을 제공합니다. 이는 프로듀서가 메시지를 트랜잭션 단위로 전송하고, 컨슈머가 'read_commited' 모드로 메시지를 읽음으로서 실현됩니다."}},{"id":"lYiCfiE149","type":"paragraph","data":{"text":"그러나 이러한 기능은 브로커 자체의 상황을 직접적으로 다루는 것이 아니라, 메시지 전송과 소비 과정에서의 중복을 방지하는데 초점을 맞추고 있습니다."}},{"id":"cwUyVYIY2F","type":"paragraph","data":{"text":"브로커 장애가 발생한 경우, kafka 클러스터의 내결함성 매커니즘인 복제 (replication) 와 리더 선출 (leader election) 프로세스가 중요해집니다. 클러스터 내에 충분한 복제본이 유지되고, 장애가 발생한 브로커의 파티션에 대해 새로운 리더가 선출될 수 있다면, 데이터 손실 없이 메시지 처리가 개속될 수 있습니다."}},{"id":"LuBxju-43o","type":"delimiter","data":{}},{"id":"OmRA2tVC70","type":"customHeader","data":{"text":"최신 Offset 으로 재시작 with RDBMS"}},{"id":"rBJVpmbYk8","type":"paragraph","data":{"text":"Consumer 가 처리한 메시지의 오프셋을 정확히 관리하고, 중복 처리 상황을 완벽하게 해결할 수 없는 경우가 있습니다. 예를 들어, Consumer 가 처리한 메시지의 오프셋을 관리하는 과정에서 장애가 발생하거나, 컨슈머 내부에서의 재처리 로직이 복잡한 경우 외부 시스템을 통한 중복 검사가 필요할 수 있습니다."}},{"id":"18Q5i_8hNm","type":"header","data":{"text":"- 오프셋 처리 상태 저장을 위한 테이블 설계","level":3}},{"id":"LjfCDznFN3","type":"code","data":{"code":"CREATE TABLE kafka_offsets (\n partition_id INT NOT NULL,\n offset BIGINT NOT NULL,\n status VARCHAR(20),\n processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n PRIMARY KEY (partition_id, offset)\n);"}},{"id":"Lez0cBD6xz","type":"header","data":{"text":"- 메시지 처리 후 오프셋과 상태 저장 로직 ","level":3}},{"id":"uoDfUwRZmJ","type":"paragraph","data":{"text":"Kafka 컨슈머가 메시지를 처리한 후, 해당 메시지의 파티션 ID, 오프셋, 처리 상태를 데이터베이스에 저장합니다. 이때, 처리 성공 여부에 따라 상태를 'PROCESSED' 또는 'FAILED' 로 표시할 수 있습니다."}},{"id":"vTPyxBU6Ra","type":"code","data":{"code":"// 예시 Java 코드\npublic void saveOffsetAndStatus(int partitionId, long offset, String status) {\n // JDBC를 사용하여 데이터베이스 연결 및 SQL 실행\n String sql = \"INSERT INTO kafka_offsets (partition_id, offset, status) VALUES (?, ?, ?)\";\n // PreparedStatement와 함께 sql을 실행하는 로직 구현\n}"}},{"id":"y54TzE9vbG","type":"header","data":{"text":"- 특정 오프셋에서 재처리 시작 로직 ","level":3}},{"id":"VZnUyJDE-J","type":"paragraph","data":{"text":"Kafka 브로커 장애 후, 최신 오프셋에서부터 재처리하기 위해서는, 컨슈머가 시작될 때 데이터베이스에서 각 파티션의 최신 오프셋을 조회하고, 그 오프셋부터 메시지를 소비하도록 Kafka 컨슈머의 오프셋을 설정해야 합니다."}},{"id":"RqbZdE4a7o","type":"code","data":{"code":"// 예시 Java 코드\npublic Map getLatestOffsets() {\n Map latestOffsets = new HashMap<>();\n // JDBC를 사용하여 \"SELECT partition_id, MAX(offset) FROM kafka_offsets GROUP BY partition_id\" 쿼리 실행\n // 결과를 latestOffsets에 저장\n return latestOffsets;\n}\n\npublic void seekToLatestOffsets(KafkaConsumer consumer, String topic) {\n Map latestOffsets = getLatestOffsets();\n // 각 파티션의 최신 오프셋으로 컨슈머의 오프셋 조정\n latestOffsets.forEach((partitionId, offset) -> {\n TopicPartition topicPartition = new TopicPartition(topic, partitionId);\n consumer.assign(Collections.singletonList(topicPartition));\n consumer.seek(topicPartition, offset);\n });\n}"}},{"id":"w02bH26xgK","type":"delimiter","data":{}},{"id":"1aIhLOIESx","type":"customHeader","data":{"text":"특정 Offset 되돌리기 with RDBMS"}},{"id":"OJqIli2voE","type":"paragraph","data":{"text":"특정 오프셋부터 다시 시작하면서 그 이후에 처리된 이벤트를 건너뛰고 싶은 경우, 처리 로직을 세심하게 관리하여 이미 처리된 이벤트를 식별하고 건너뛸 수 있도록 해야합니다."}},{"id":"ecHVLv3HJL","type":"paragraph","data":{"text":"하지만 이 방식은 RDBMS 의 빈번한 통신이 추가되므로 성능에 미치는 영향을 고려하여 적절한 최적화 및 캐싱 전략을 적용해야 합니다."}},{"id":"lRzSjrGZ8m","type":"header","data":{"text":"- 이벤트 처리 상태 확인","level":3}},{"id":"l63698hxxa","type":"paragraph","data":{"text":"이벤트를 처리하기 전에, 해당 이벤트의 'partition_id' 와 'offset' 을 사용하여 'kafka_offsets' 테이블에서 이벤트의 처리 상태를 조회합니다. 만약 'status' 가 'PROCCESSED' 라면 이미 처리된 이벤트 이므로 이를 건너뛰고 다음 이벤트로 넘어갑니다."}},{"id":"cWt0S13Abm","type":"code","data":{"code":"public boolean checkIfProcessed(int partitionId, long offset) {\n // JDBC를 사용하여 데이터베이스에서 해당 partition_id와 offset에 대한 레코드 조회\n // 조회 결과에서 status가 'PROCESSED'인지 확인\n // 'PROCESSED'인 경우 true 반환, 그렇지 않은 경우 false 반환\n}\n"}},{"id":"SrOLmgkxzk","type":"header","data":{"text":"- 이벤트 처리 및 상태 업데이트","level":3}},{"id":"6kmAF1QXPN","type":"paragraph","data":{"text":"이벤트가 아직 처리되지 않았다면 ('status' 가 'PROCCESSED' 가 아니면), 이벤트 처리 로직을 수행한 후, 처리결과에 따라 'kafka_offsets' 테이블에 해당 이벤트의 상태를 'PROCESSED' 로 업데이트 합니다."}},{"id":"tk5P3ZA4IX","type":"code","data":{"code":"public void processEvent(int partitionId, long offset, String eventData) {\n if (!checkIfProcessed(partitionId, offset)) {\n // 이벤트 처리 로직 수행\n \n // 처리가 완료된 후, 데이터베이스에 해당 이벤트의 상태를 'PROCESSED'로 업데이트\n updateEventStatus(partitionId, offset, \"PROCESSED\");\n }\n}"}},{"id":"7tIJQSFv9G","type":"header","data":{"text":"- 이벤트 상태 업데이트","level":3}},{"id":"DMgE5h1m71","type":"paragraph","data":{"text":"이벤트 처리 후, 'kafka_offsets' 테이블에 해당 이벤트의 상태를 'PROCESSED' 로 업데이트 하는 로직을 구현합니다."}},{"id":"_kHCq46Vln","type":"code","data":{"code":"public void updateEventStatus(int partitionId, long offset, String status) {\n // JDBC를 사용하여 데이터베이스에 연결\n // UPDATE 쿼리를 실행하여 해당 partition_id와 offset의 레코드의 status를 업데이트\n}"}},{"id":"Pwf76N05UD","type":"paragraph","data":{"text":"이와 같은 방식으로 'kafka_offsets' 테이블에 저장된 상태정보를 기반으로 이미 처리된 이벤트를 효과적으로 건너뛸 수 있으며, 중복 처리를 방지할 수 있습니다."}},{"id":"7tw8kvPagw","type":"customLink","data":{"url":"https://medium.com/event-driven-utopia/understanding-kafka-topic-partitions-ae40f80552e8","title":"Understanding Kafka Topic Partitions | by Dunith Danushka | Tributary Data | Medium","description":"Everything in Kafka is modeled around partitions. They rule Kafka's message storage, scalability, replication, and message movement.","image":"https://miro.medium.com/v2/da:true/resize:fit:1200/0*S54oesPGxCXI3_9k"}}],"version":"2.31.0-rc.7"}

Tag List