카프카 컨슈머 Kafka Consumer는 사용자가 제어하며 poll 메소드를 호출하여 카프카 브로커에서 데이터를 가져올 수 있다. 일반적으로 Consumer의 group.id와 가져올 topic을 지정한다. 같은 group.id를 사용하는 Consumer를 묶어서 Consumer Group이라고 한다. // KafkaConsumer ... set properties of consumer val consumer = consumerFactory(props).createConsumer() consumer.subscribe(listOf("topic-name")) while (true) { val records = consumer.poll(Duration.ofMillis(100)) records.forEach ..
카프카 프로듀서 프로듀서는 카프카 프로듀서 API를 포함하여 구성된 애플리케이션을 마라혀 브로커에 특정 토픽의 파티션에 메시지를 전달하는 역할을 합니다. 프로듀서가 전달하는 메시지의 구조는 다음과 같습니다. Topic (토픽) Partition (특정 파티션 위치) Timestamp (생성 시간) Header (헤더) Key (키) Value (값) 카프카 서버에 메시지를 전송하는 책임을 가지고 있으며 데이터를 전송하기 위한 전략을 가지고 있다. 재시도는 어떻게 할 것 인지 데이터 전송 시 Compression은 어떻게 할 것 인지 데이터 전송 시 Serializer는 어떤것을 사용할 것인지 어떤 Partition으로 데이터 전송을 할 것인지 Compression (압축) 메시지 압축 옵션을 사용하였다면..
카프카 구조 Zookeeper (주키퍼) 분산 코디네이션 서비스를 제공하는 오픈소스 프로젝트로 직접 어플리케이션 작업을 조율하는 것을 쉽게 개발할 수 있도록 도와주는 도구이다. API를 이용해 동기화나 마스터 선출 등의 작업을 쉽게 구현할 수 있게 해준다. Broker (브로커) Kafka 서버를 의미하며 한개의 Cluster 내에 여러개의 Kafka 서버를 실행시킬 수 있다. Topic (토픽) 메시지가 생산되고 소비되는 주체 Partition (파티션) 하나의 토픽내에서 토픽의 메시지가 분산되어서 각자 저장되는 단위 (위 카프카 아키텍쳐대로면 한개의 토픽에서 3개의 파티션으로 저장 관리되고 있다.) 파티션은 내부가 Queue로 구성이 되어있어서 순서를 보장하지만, Partition끼리의 순서를 보장하..
Pattern 1: Stop on error (에러시 중지) 모든 이벤트들이 순서 보장과 함께 에러 없이 처리 되어야 하는 경우에 사용함. (예를 들어, CDC) 처리 도중 에러가 발생한다면 어플리케이션은 중단되며 수동 개입이 필수적이다. Source 토픽의 이벤트는 다른 경로를 사용하지 않는다. Pattern 2: Dead letter queue (실패 대기열 큐) 일반적인 시나리오로 메인 스트림이 계속되는 동안 어플리케이션에서 처리가 불가하다면 Error 토픽으로 향하도록 함. 이 방식에서는 재시도를 위한 프로세스를 요구하거나 지원하지 않는다. 즉, 이벤트는 성공적으로 처리되거나 Error 토픽이다. 일반적인 상황에서 어플리케이션은 Source 토픽의 각 이벤트를 처리하고 Target 토픽으로 생성한..
[Kafka] Python confluent Kafka 설치 및 테스트 테스트 및 설치 환경은 Mac에서 진행 하였습니다. Zookeper와 Kafka Cluster docker-compose.yml version: '3' services: zookeeper: image: confluentinc/cp-zookeeper:7.3.0 hostname: zookeeper container_name: zookeeper environment: TZ: Asia/Seoul ZOOKEEPER_SERVER_ID: 1 ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 kafka-1: image: confluentinc/cp-kafka:7.3.0 container_..
