DevBoi

프로듀서 에서는 파티셔너를 제일 많이 쓴다. 총 2개의 파티셔너에 대한 지원을 해준다. UniformStickyPartitioner, RoundRobinPartitioner 2개의 파티셔너를 제공한다. 카프카 클라이언트 라이브러리에서 파티셔너를 지정하지 않는 경우, UniformStickyPartioner가 파티셔너로 기본 설정된다. 1. 메시지 키가 있을 경우 동작 UniformStickyPartioner와 RoundRobinPartitioner 둘다 메시지 키가 있을때는 메시지 키의 해시값과 파티션을 매칭하여 레코드를 전송한다. 동일한 메시지키가 존재하는 레코드는 동일한 파티션 번호에 전달 된다. 만약 파티션 개수가 변경되는 경우,메시지키와 파티션 번호 매칭은 깨지게 된다. 즉 메시지키가 있는 경우,..

카프카 토픽을 사용할때 가장많이 사용하는 도구이다. 프로듀서 애플리케이션을 개발해서, 토픽에 메시지를 발행하고 일반적인 환경에서는 Producer 애플리케이션을 개발해서 토픽에 메시지를 넣는다. 카프카의 데이터는 시작점이 프로듀서이다. 프로듀서 애플리케이션은 카프카에 필요한 데이터를 선언하고 브로커의 특정 토픽의 파티션에 전송한다. 프로듀서는 리더 파티션이 있는 브로커와 통신하게 된다. 즉 프로듀서는 토픽의 리더파티션을 가지고있는 브로커와 통신하고 다른 팔로워 파티션은 이를 복제한다. 또한 이런 애플리케이션이 자바가 아닌 라이브러리를 사용하면, 공식적으로 지원하는 라이브러리가 아니기떄문에 문제가 발생할 수 있다. 따라서 자바로 개발하는게 좋다. 파티셔너,배치생성 단계를 거치고 데이터를 브로커로 전송하게 ..

dead letter queue는 이름에서 유추 할 수 있듯이 실패한 레코드를 보관하는 별도의 큐이다. 이름에서 알 수 있듯이. 실패한 레코드를 보관하는 별도의 큐이다. 카프카는 2.0 부터 자체 plug in으로 실패한 레코드의 메타정보도 포함 시켜 저장하는 기능을 제공한다. 설정 방법은 간단하다. Connect 설정에 아래와 같이 추가해주면 된다. connect_dlq에서 아래와 같이 설정한다. errors.tolerance = all errors.deadletterqueue.topic.name = {토픽} // 유효하지 않다고 판단된 레코드의 메타데이터 header에 저장 errors.deadletterqueue.context.headers.enable = true // 로그 파일에 유효하지 않는 ..

토픽을 생성하는 상황은 크게 2가지이다. 1. 컨슈머 or 프로듀서가 카프카 브로커에 생성되지 않은 토픽에 대해 데이터를 요청할때 2. 커맨드 라인툴로 명시적으로 토픽을 생성하는 것이다. 유지보수를 위해서는 명시적으로 생성하는 것이좋지만 동시 처리량이 많아야 하는 토픽의 경우 파티션의 개수를 다르게 설정할 수 있고 단기간 데이터 처리만 필요한 경우 토픽에 들어온 데이터의 보관기간 옵션을 짧게 설정할 수 있다. 토픽의 특성별로 다르게 관리가 되어야 하기 때문에 명시적으로 생성하는 것이 좋다. broker의 옵션 중 토픽 auto topiccreate true로 주게 되면? 컨슈머나 프로듀서의 요청이 지금 없는 토픽이라면? 신규 토픽에 대해서 토픽을 생성해주는 것이다. (1번상황)

kafka-consumer-perf-test.sh는 컨슈머로 퍼포먼스를 측정할 때 사용된다. 카프카 브로커와 컨슈머간의 네트워크를 체크할 때 사용한다. kafka-reassign-partitions.sh 해당 과 같이 브로커 마다 파티션이 몰릴 수 있다. 이때 리더파티션과 팔로워 파티션을 적절하게 분산할 수 있는데, 해당 분산시에 사용된다. 근데 이걸 사용하면 리더 파티션과 팔로워 파티션이 위치를 변경 할 수 있다. kafka broker에는 auto leader rebalance.enable옵션이 있는데, 이 옵션의 기본값은 true로 클러스터 단위에서 리더 파티션을 자동 리밸런싱 하도록 도와준다. 브로커의 백그라운드 스레드가 일정한 간격으로 리더의 위치를 파악하고 필요시 리더 리밸런싱을 통해 리더의 ..

컨슈머 그룹에 대한 걸 알아보자. 컨슈머 그룹으로 생성된 컨슈머로 해당 토픽에 대한 데이터를 가져갔다. 컨슈머 그룹은 따로 생성하는 명령을 날리지 않고 컨슈머를 동작할 때 컨슈머 그룹이름을 지정하면 새로 생성된다. 또한 어떤 컨슈머 그룹이 어떤 토픽을 대상으로 레코드를 가져갔는지 그리고 상태와 오프셋 컨슈머 렉 컨슈머 ID,호스트 등을 알 수 있다. (컨슈머 렉 - 현재 레코드의 오프셋과 마지막 레코드의 차이이다 ,즉 컨슈머가 읽지않은 데이터라고 이해하면 된다.) 컨슈머의 상태를 조회할때도 유용하다. 또한 컨슈머그룹의 오프셋을 리셋할 수 있다. 오프셋의 리셋은 처음부터 시작되는 것만은 아니다. 특정 오프셋으로 이동할 수 있는 것이다. 순차처리가 필요없는 경우, 최신의 데이터로 오프셋을 이동할 수 있다는 ..

카프카의 컨슈머 역할을 한다. 토픽에 대한 조회 용도로 사용한다. 특정 카프카 브로커의 토픽의 데이터를 가져올 수 잇다. bootstrap server, topic에 대한 옵션이 필수이고, 저장된 데이터를 가장 처음 부터 출력이 가능하다. 토픽의 데이터도 가져올 수 있고, 메시지키를 가져올 수도 있다. 최대 컨슘 메시지 개수를 설정할 수도있고 특정 파티션에 대한 컨슘도 가능하다. 진행해보자 1) 주키퍼, 브로커 데몬 모드로 기동 bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties 2) 토픽 생성 bin/kafka-topics.sh --..

hello.kafka 토픽에 데이터를 넣을수 있는 명령어를 실행해보자 키보드로 문자를 작성하고 엔터키를 누르면 별다른 응답없이 메시지 값이 전송된다. kafka-console-producer.sh 쉡 스크립트를 사용하면 된다. 그리고, 메시지를 보낼때 separator를 선언하면, 메시지키를 구분자로 사용할 수 있다. 메시지 키를 동일하게 쓰면, 같은 파티션으로 들어가게 된다. 즉 동일한 메시지키를 쓰면, 같은 파티션으로 들어가게 된다. 메시지 키가 Null인 경우, 라운드 로빈으로 전송한다. 즉 여러 파티션에 들어가게 된다. 실습 해보자 bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka 이렇게 하는 경우 ..

kafka-configs 를 정리해보자 토픽의 일부 옵션을 설정하기 위해서는 kafka-configs.sh명령어를 사용해야한다. --alter와 --add-config 옵션을 사용하여, Min.insync.replicas옵션을 토픽별로 설정할 수있다. 관련 Config 명령어로 broker,all,describe옵션을 사용해서 조회가 가능하다. bin/kafka-configs.sh --bootstrap-server my-kafka:9092 --alter --add-config min.insync.replica=2 topic test bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --describe 예를 들어 위와 같이, ..

토픽은 디비의 테이블과 같다. 토픽을 생성하려면, 클러스터 정보와 토픽의 이름을 알고 있으면 된다. topic.sh를 활용해서 해당 토픽을 생성하거나,컨트롤 하면 된다. 파티션 개수, 복제 개수를 설정할 수 있지만, 안하면 브로커의 기본값으로 세팅된다. 우선 주키퍼와 브로커를 실행하자 실행파일은 bin 하위, config 파일은 config 폴더 하위에 있다. 파티션 개수를 줄일수 없는 특성 때문에, 해당 파티션의 개수를 줄여야 할 때는 방법이 없다. 이때는 그냥 새로 토픽을 만드는 것이 좋다. 일단. 주키퍼, 브로커를 실행하자 1) 주키퍼 실행 bin/zookeeper-server-start.sh config/zookeeper.properties 2) 브로커 실행 bin/kafka-server-star..