KAFKA

데이터를 생성하는 소스 애플리케이션과 최종적으로 데이터가 적재되는 타깃 애플리케이션의 연결을 위해 탄생소스 어플리케이션과 타깃 어플리케이션의 개수가 많아짐에 따라 파이프라인이 복잡해짐각각의 어플리케이션을 따로따로 연결하는 것이 아닌 카프카를 통해 한 곳에 데이터를 모아

카프카 생태계를 살펴보면 위와 같다.프로듀서, 컨슈머, 스트림즈, 커넥트는 모두 오픈소스 아파치 카프카에 기본적으로 릴리즈되어 있는 툴이다.카프카 클러스터가 가운데에 위치해 있는데 목적에 맞게 만들어진 토픽을 가지고 있다.프로듀서가 데이터를 전송하면 카프카 클러스터의

주키퍼는 카프카 2.xx 버전까지는 카프카 클러스터를 운영하기 위해 반드시 필요한 애플리케이션이다.카프카 3.xx 버전부터는 주키퍼가 없이도 카프카를 운영할 수 있으나, 아직까지 완벽하게 대체하지 못하고 있기 때문에 상용환경에서는 카프카 클러스터 운영 시 주키퍼를 함께

kafka-topics.sh 명령어를 이용해 토픽 생성 시 다음과 같은 오류가 났다.부트스트랩 서버 정보를 잘못 입력해서 발생한 오류였다.일단 hosts 설정에 kakfa라는 설정이 없으니, 찾지 못한 것이다.my-kakfa -> my-kafka로 수정하고 실행 시 잘

카프카 커맨드 라인 툴(Command-line Tool)은 카프카를 운영할 때 브로커 운영에 필요한 다양한 명령을 내릴 수 있는 도구이다.데이터를 주고 받는 작업 뿐 아니라, 카프카 토픽이나 파티션 개수 변경과 같은 운영을 할 필요가 있기 때문에 카프카 커맨드 라인툴을

bin 디렉토리 하위 kafka-topics.sh 라는 이름의 쉘 스크립트가 있다.kafka-topics.sh는 토픽과 관련된 쉘 스크립트 커맨드 라인 툴이다.토픽 생성에는 create를 사용한다.필수값인 카프카 클러스터 정보와 토픽 이름을 지정해 토픽을 생성할 수 있

bin 디렉토리 하위 kafka-configs.sh 라는 이름의 쉘 스크립트가 있다.kafka-configs.sh는 토픽의 옵션 설정과 관련된 쉘 스크립트 커맨드 라인 툴이다.alter 옵션과 add-config 옵션을 사용해 min.insync.replicas 옵션을

bin 디렉토리 하위 kafka-producer-perf-test.sh 라는 이름의 쉘 스크립트가 있다.kafka-producer-perf-test.sh는 카프카 프로듀서로 퍼포먼스를 측정할 때 사용한다.

카프카에서 데이터의 시작점은 바로 프로듀서이다. 카프카 프로듀서 애플리케이션의 역할은 카프카에 필요한 데이터를 선언하고 브로커의 특정 토픽의 특정 파티션에 전송해준다.프로듀서는 데이터를 전송할 때 리더 파티션을 가지고 있는 카프카 브로커와 직접 통신한다.프로듀서는 카프

build.gradle

저번 포스팅과 같이 프로듀서레코드를 생성할 때 토픽명과 메시지값만을 설정할 수 있다.이 때는 생성자의 파라미터로 1. 토픽명, 2. 메시지값 을 설정해준다.만약 키를 함께 보내고 싶을 경우생성자의 파라미터로 1. 토픽명, 2. 메시지키 3. 메시지값 을 설정해준다.Pr

레코드는 기본적으로 해시값에 의해 자동으로 파티션에 분배된다.그러나 운영하다보면 특정 데이터의 레코드를 특정 파티션으로 전송해야 하는 경우가 있다.저번 포스팅과 같이 레코드를 생성할 때 레코드에 파티션 번호를 파라미터로 지정해주는 방법이 있다.이렇게 각각의 레코드에 직

KafkaProducer.java KafkaProducer의 send() 메소드는 Future 객체를 반환한다. 이 Future 객체는 RecordMetadata의 비동기 결과를 표현하고, ProducerRecord가 카프카 브로커에 정상 적재되어있는지에 대한 정보가

프로듀서가 전송한 데이터는 카프카 브로커의 토픽에 적재된다.컨슈머는 적재된 데이터를 사용하기 위해 브로커로부터 데이터를 가져와서 필요한 처리를 한다.컨슈머는 데이터를 가져와서 데이터베이스에 저장하거나 다른 애플리케이션과 통신하는 역할을 한다.

컨슈머 애플리케이션을 개발해보자.프로듀서 애플리케이션과 동일하게 build.gradle 디펜던시 추가가 필요하다.아파치 카프카를 배포할 때 사용하는 공식 라이브러리인 kafka-clients와 로그를 남기기 위한 slf4j 디펜던시를 추가한다.

앞선 포스팅에서는 자동 커밋을 수행하는 컨슈머 애플리케이션을 개발해보았다. 이번에는 수동 커밋을 수행하는 컨슈머 애플리케이션을 개발해볼 것 이다. 수동 커밋의 종류는 동기 커밋, 비동기 커밋으로 두가지가 있다. 1. 수동 커밋 동기 커밋 2.

리밸랜스가 자주 발생하면 안되긴 하지만, 리밸랜스가 발생할 경우 리스너를 통해 대응 로직을 짤 수 있다.