$$config title: [kafka] 카프카 완전 정복 (confluent) keywords: - IT - ... $$ # confluent 라이센스 - [https://www.confluent.io/installation/](https://www.confluent.io/installation/) ``` Local (무료) Distributed (유료) Community (무료) ``` # 1. Local Kafka ## (1) 카프카 local install - [https://packages.confluent.io/archive/](https://packages.confluent.io/archive/) - `confluent 7.1.2` 버전으로 진행 ### A. 다운로드 ```bash wget https://packages.confluent.io/archive/7.1/confluent-7.1.2.tar.gz ``` ### B. 압축풀기 ```bash tar -zxvf confluent-7.1.2.tar.gz ``` ## (2) openjdk 11 설치 ### A. 다운로드 ```bash wget https://download.java.net/openjdk/jdk11/ri/openjdk-11+28_linux-x64_bin.tar.gz ``` ### B. 압축풀기 ```bash tar zxvf openjdk-11+28_linux-x64_bin.tar.gz ``` ### C. /usr/local 로 이동 ```bash sudo mv jdk-11* /usr/local/ ``` ### D. 환경 변수 등록 ```bash vi ~/.bashrc ``` - 맨 아래에 추가 ```bash export JAVA_HOME=/usr/local/jdk-11 export PATH=.:$PATH:$JAVA_HOME/bin ``` - 적용 ``` cd ~ . .bashrc ``` - 확인 ```bash echo $JAVA_HOME ``` ## (3) confluent 경로 설정 ```bash vi ~/.bashrc ``` - 맨 아래에 추가 ```bash export CONFLUENT_HOME=/kafka/confluent export PATH=.:$PATH:$CONFLUENT_HOME/bin ``` - 적용 ``` cd ~ . .bashrc ``` - 확인 ```bash echo $CONFLUENT_HOME ``` ## (4) 주키퍼 기동 ```bash zookeeper-server-start $CONFLUENT_HOME/etc/kafka/zookeeper.properties ``` > zookeeper.properties 에서 log.dir이 /tmp 경로 일 경우 vm이 다시 기동되면 데이터 사라짐 ## (5) 카프카 기동 ```bash kafka-server-start $CONFLUENT_HOME/etc/kafka/server.properties ``` > zookeeper.properties 에서 log.dir이 /tmp 경로 일 경우 vm이 다시 기동되면 데이터 사라짐 ## (6) 토픽 생성 - `welcome-topic`이라는 토픽의 이름으로 토픽을 생성 - 파티션의 개수는 기본 default는 server.properties에 num.partitions=1로 설정되어 있기 때문에 따로 설정을 해주지 않으면 1개가 생성된다 ```bash kafka-topics --bootstrap-server localhost:9092 --create --topic welcome-topic ``` - 명시적으로 파티션의 개수를 지정하여 토픽을 생성 ```bash kafka-topics --bootstrap-server localhost:9092 --create --topic test_topic_02 --partitions 3 ``` ## (7) replicatin-factor=2 - 파티션은 독립적으로 존재하게 되는데 만약 파티션이 10개가 있다면 그 중에 하나만 문제가 발생해도 데이터의 손실이 발생하게 된다 - 그래서 파티션을 복제하여 파티션이 사라지더라도 복제한 파티션이 있기 때문에 데이터의 손실을 방지할 수 있다. - `replication-factor=2`라는 설정은 각각 파티션의 복제 갯수가 1개이다(원본 + 복제) - 결론적으로 원본과 복제1개가 있다는 뜻이다. > 추가로 replication-factor는 노드의 개수 만큼 지정할 수 있다. > 단일 노드일 경우 replication-factor를 사용할 수 없다 ## (8) WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both. - topic을 생성할 때 `.`과 `_`를 같이 사용하면 충돌이 발생할 수 있다 - 그러므로 topic을 생성할 때 같이 사용하지 말도록 하자 ## (9) ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 2 larger than available brokers: 1. - `replication-factor` 옵션은 브로커의 개수 만큼 설정이 가능하므로 단일 노드의 경우에서 2개를 설정할 때 에러가 발생할 수 있는걸 볼 수 있다. ## (10) 토픽 상세보기 ```bash kafka-topics --bootstrap-server localhost:9092 --describe --topic test_topic_02 ``` - 총 3개의 파티션이 생성이 됨 - 단일 노드이므로 replication factor는 1 ```bash Topic: test_topic_02 TopicId: JNfSg1l1QpKTY7BN3CsRYg PartitionCount: 3 ReplicationFactor: 1 Configs: segment.bytes=1073741824 Topic: test_topic_02 Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: test_topic_02 Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: test_topic_02 Partition: 2 Leader: 0 Replicas: 0 Isr: 0 Offline: ``` ## (11) 토픽 삭제 ```bash kafka-topics --bootstrap-server localhost:9092 --delete --topic test_topic_02 ``` - 토픽을 삭제하였을 때 바로 삭제가 되지 않고 `.6a95e0cd1a5a461f93f3c27114d1a710-delete`가 붙게 된다 - 1분정도의 시간이 소요되고 삭제가 된다. ```bash test_topic_02-0.6a95e0cd1a5a461f93f3c27114d1a710-delete test_topic_02-1.01628a21c55c40828cfcbafdae11a5bf-delete test_topic_02-2.9c6c85c4490a422786f8b6fdecbfeb75-delete ``` > 카프카의 `server.properties`의 파일에서 `log.dirs`의 경로를 가보면 볼 수 있다. ## (12) Record = Message = Event - 모두 같은 의미를 가진다 - 레코드의 구조 ``` topic partition key value headers ``` ## (13) consumer의 auto.offset.reset - default로 `latest`로 되어있어 offset중에 가장 최신의 데이터부터 가져온다 - 그러나 `kafka-console-consumer`에서 `--from-beginning`의 옵션을 붙여주면 `earliest`가 설정이 되어 맨 처음부터 데이터를 가져온다 ## (14) producer console 실행 ```bash kafka-console-producer --bootstrap-server localhost:9092 --topic test-topic ``` ## (15) consumer console 실행 ```bash kafka-console-consumer --bootstrap-server localhost:9092 --topic test-topic --from-beginning ``` ## (16) 카프카의 전송원리 중 serializer deserializer - 자바 기반으로 되어있기 때문에 자바의 기능을 사용 - 카프카는 byte[]만 받는다 그래서 객체를 byte[]로 변환해주어야하고 또 byte[]를 객체로 변환해주어야한다. ``` object =(serializer)> byte[ ] => kafka => byte[ ] =(deserializer)> object ``` ## (17) key가 있는 레코드 전송 ```bash kafka-console-producer --bootstrap-server localhost:9092 --topic test-topic \ --property key.separator=: --property parse.key=true ``` ```bash kafka-console-producer --bootstrap-server localhost:9092 --topic multipart-topic \ --property key.separator=: --property parse.key=true ``` ## (18) key가 있는 레코드 받기 ```bash kafka-console-consumer --bootstrap-server localhost:9092 --topic test-topic \ --property print.key=true --property print.value=true --from-beginning ``` ```bash kafka-console-consumer --bootstrap-server localhost:9092 --topic multipart-topic \ --property print.partition=true --property print.key=true --property print.value=true \ --from-beginning ``` ## (19) key가 있는 레코드는 항상 같은 파티션에 간다 (같은 파티션은 순서보장, 다른 파티션은 순서보장 안됨) - 컨슈머가 해당 토픽을 받을 때는 무작위로 파티션을 선택하여 받는다 - 그러나 같은 key를 가진 레코드는 항상 같은 파티션으로 가게되어 순서가 보장이 된다 ``` >1:aaa >2:bbb >3:ccc >4:ddd >5:eee >6:fff >1:kkk >2:bbbaaa ``` ``` Partition:2 null a Partition:2 null ccc Partition:2 null eee Partition:2 null hhh Partition:0 null ddd Partition:0 null hhh Partition:1 null bbb Partition:1 null ggg Partition:0 1 aaa Partition:2 2 bbb Partition:2 3 ccc Partition:1 4 ddd Partition:0 5 eee Partition:1 6 fff Partition:0 1 kkk Partition:2 2 bbbaaa ``` > 보낸 순서와 다르게 컨슈머는 받는다 그러나 같은 파티션에 있는 레코드들은 순서를 보장 받는다. - 결론은 같은 키가 있으면 항상 같은 파티션으로 간다 - 그리고 같은 파티션에 있는 레코드는 순서를 보장 받을 수 있다. ## (20) 모든 컨슈머는 한 컨슈머그룹에 속해있다 - 모든 컨슈머는 어떤 특정의 한개의 컨슈머 그룹에 속해있다 - 그리고 그 컨슈머 그룹은 독립적으로 존재한다 ## (21) 하나의 파티션은 하나의 컨슈머만 할당받을 수 있다 - 파티션을 한 컨슈머 그룹안에서 다수의 컨슈머에게 할당을 받게 된다면 파티션의 순서보장을 할 수 없으므로 하나의 파티션은 하나의 컨슈머에 할당이 된다 ## (22) 컨슈머 그룹을 지정하여 컨슈머 실행 ```bash kafka-console-consumer --bootstrap-server localhost:9092 --group group_01 --topic multipart-topic \ --property print.key=true --property print.value=true \ --property print.partition=true ``` ## (23) 컨슈머 그룹 자세히 보기 ```bash kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group group_01 ``` - 결과 ``` Consumer group 'group_01' has no active members. GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID group_01 multipart-topic 0 0 692 692 - - - group_01 multipart-topic 1 0 656 656 - - - group_01 multipart-topic 2 0 656 656 - - - ``` - 지금은 컨슈머를 할당했다가 취소해서 `CONSUMER-ID`에 컨슈머가 할당이 되어있지 않다 - `LAG`는 밀린 레코드의 개수이다 - 해당 명령어를 통해 컨슈머의 리밸런싱을 볼 수 있다. - 컨슈머그룹은 default 7일정도 머물러 있다가 삭제된다. ## (24) 컨슈머 그룹 삭제 ```bash kafka-consumer-groups --bootstrap-server localhost:9092 --delete --group group_01 ``` > 단 컨슈머 그룹에 속해 있는 모든 컨슈머가 다운되어있지 않다면 컨슈머그룹을 삭제할 수 없다. ## (25) 해당 토픽의 데이터가 물리적으로 잘 들어가 있는지 확인 ```bash kafka-dump-log --deep-iteration --files /root/data/kafka-logs/multipart-topic-0/00000000000000000000.log --print-data-log ``` # 2. JAVA kafka ## jdk version - jdk 버전은 17로 진행 ## 의존성 설치 ```groovy dependencies { implementation 'org.apache.kafka:kafka-clients:3.1.2' implementation 'org.slf4j:slf4j-api:2.0.3' implementation 'org.slf4j:slf4j-simple:2.0.3' implementation 'com.github.javafaker:javafaker:1.0.2' } ``` ## (1) Producer ### A. simple producer - `BOOTSTRAP_SERVERS_CONFIG`의 값은 원래 `192.168.56.102:9092`이다 - 그러나 `java.unkownhost exception`이 발생하면 `vi /etc/hosts`에서 해당 아이피를 등록하여 `ubuntu2:9092`로 작성하면 된다 - ```java package org.example.kafka; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class SimpleProducer { public static void main(String[] args) { String topicName = "simple-topic"; // KafkaProducer Configuration // null, "hello world" Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); // producer record object creation ProducerRecord producerRecord = new ProducerRecord<>(topicName, "hello world"); // kafka producer message send // send시 kafka에 전송이 되지 않고 flush가 발생했을 때 kafka에 전송이 된다. (배치와 연관이 있음) kafkaProducer.send(producerRecord); kafkaProducer.flush(); kafkaProducer.close(); } } ``` ### B. simpler producer sync - 기본적으로 producer send는 비동기로 작동하여 `future<>`을 반환하지만 - 뒤에 `get()`을 붙이게 되면 동기로 작동하고 `RecordMetadata` 객체를 반환한다. ```java package org.example.kafka; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; import java.util.concurrent.ExecutionException; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class SimpleProducerSync { public static final Logger logger = LoggerFactory.getLogger(SimpleProducerSync.class.getName()); public static void main(String[] args) { String topicName = "simple-topic"; // KafkaProducer Configuration // null, "hello world" Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); // producer record object creation ProducerRecord producerRecord = new ProducerRecord<>(topicName, "hello world"); // kafka producer message send try { RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get(); logger.info("\n ####### record metadata received ######## \n" + "partition: " + recordMetadata.partition() + "\n" + "offset: " + recordMetadata.offset() + "\n" + "timestamp: " + recordMetadata.timestamp()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } finally { kafkaProducer.close(); } } } ``` ### C. simpler producer ASync - producer가 비동기로 진행이 되며 send 메서드의 두번째 매개변수에 callback을 작성하여 해당 send 메서드가 비동기로 진행이 되고 카프카에서 응답이 오면 callback 메서드를 호출합니다. - `Thread.sleep(3000)`을 주었기 때문에 비동기여도 3초라는 시간을 스레드를 주었기 때문에 카프카에서 응답이 와서 callback이 호출됩니다. ```java package org.example.kafka; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class SimpleProducerASync { public static final Logger logger = LoggerFactory.getLogger(SimpleProducerASync.class.getName()); public static void main(String[] args) { String topicName = "simple-topic"; // KafkaProducer Configuration // null, "hello world" Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); // producer record object creation ProducerRecord producerRecord = new ProducerRecord<>(topicName, "hello world"); // kafka producer message send kafkaProducer.send(producerRecord, (recordMetadata, exception) -> { if (exception == null) { logger.info("\n ####### record metadata received ######## \n" + "partition: " + recordMetadata.partition() + "\n" + "offset: " + recordMetadata.offset() + "\n" + "timestamp: " + recordMetadata.timestamp()); } else { logger.error("exception error from broker " + exception.getMessage()); } }); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } kafkaProducer.close(); } } ``` ### D. ProducerASyncWithKey - key가 있는 레코드를 전송하는 producer - `multipart-topic`은 파티션을 3개로 정하였습니다. - 컨슈머에서는 파티션끼리는 순서보장을 하지 않으므로 0~19까지 순차적으로 처리하지 않습니다. ```java package org.example.kafka; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class ProducerASyncWithKey { public static final Logger logger = LoggerFactory.getLogger(ProducerASyncWithKey.class.getName()); public static void main(String[] args) { String topicName = "multipart-topic"; // 3개의 파티션을 가지는 토픽 // KafkaProducer Configuration // null, "hello world" Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); for(int seq=0; seq < 20; seq++) { // producer record object creation ProducerRecord producerRecord = new ProducerRecord<>(topicName, String.valueOf(seq), "hello world %d".formatted(seq)); // kafka producer message send kafkaProducer.send(producerRecord, (recordMetadata, exception) -> { if (exception == null) { logger.info("\n ####### record metadata received ######## \n" + "partition: " + recordMetadata.partition() + "\n" + "offset: " + recordMetadata.offset() + "\n" + "timestamp: " + recordMetadata.timestamp()); } else { logger.error("exception error from broker " + exception.getMessage()); } }); } try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } kafkaProducer.close(); } } ``` ### E. ProducerASyncCustomCB - callback을 custom하여 seq의 값을 볼 수 있도록 하였습니다. - 그리고 key의 타입을 Interger로 설정했습니다. ```java package org.example.kafka; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class CustomCallback implements Callback { public static final Logger logger = LoggerFactory.getLogger(CustomCallback.class.getName()); private int seq; public CustomCallback(int seq) { this.seq = seq; } @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { logger.info("seq: {} partitions: {} offset: {}", this.seq, metadata.partition(), metadata.offset()); } else { logger.error("exception error from broker " + exception.getMessage()); } } } ``` ```java package org.example.kafka; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class ProducerASyncCustomCB { public static void main(String[] args) { String topicName = "multipart-topic"; // 3개의 파티션을 가지는 토픽 // KafkaProducer Configuration Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); for(int seq=0; seq < 20; seq++) { // producer record object creation ProducerRecord producerRecord = new ProducerRecord<>(topicName, seq, "hello world %d".formatted(seq)); // custom callback Callback callback = new CustomCallback(seq); // kafka producer message send kafkaProducer.send(producerRecord, callback); } try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } kafkaProducer.close(); } } ``` - 결과 ```bash [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 2 partitions: 2 offset: 12 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 5 partitions: 2 offset: 13 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 6 partitions: 2 offset: 14 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 12 partitions: 2 offset: 15 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 16 partitions: 2 offset: 16 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 18 partitions: 2 offset: 17 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 19 partitions: 2 offset: 18 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 1 partitions: 0 offset: 13 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 7 partitions: 0 offset: 14 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 8 partitions: 0 offset: 15 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 14 partitions: 0 offset: 16 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 15 partitions: 0 offset: 17 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 17 partitions: 0 offset: 18 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 0 partitions: 1 offset: 15 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 3 partitions: 1 offset: 16 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 4 partitions: 1 offset: 17 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 9 partitions: 1 offset: 18 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 10 partitions: 1 offset: 19 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 11 partitions: 1 offset: 20 [kafka-producer-network-thread | producer-1] INFO org.example.kafka.CustomCallback - seq: 13 partitions: 1 offset: 21 ``` > 배치형태로 send를 하는데 콜백을 받을 때도 배치형태로 받는 것 같습니다. ### F. pizza producer - pizza producer 생성하여 pizza topic의 파티션이 3개인데 컨슈머그룹을 통해 잘 분배가 되는지 확인 할 수 있는 테스트용 producer 이다 ```java package org.example.kafka; import com.github.javafaker.Faker; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Random; public class PizzaMessage { // 피자 메뉴를 설정. getRandomValueFromList()에서 임의의 피자명을 출력하는 데 사용. private static final List pizzaNames = List.of("고구마 피자", "치즈 피자", "치즈 갈릭 피자", "슈퍼 슈프림", "페페로니 피자"); // 피자 가게명을 설정. getRandomValueFromList()에서 임의의 피자 가게명을 출력하는데 사용. private static final List pizzaShop = List.of("A001", "B001", "C001", "D001", "E001", "F001", "G001", "H001", "I001", "J001", "K001", "L001", "M001", "N001", "O001", "P001", "Q001"); public PizzaMessage() {} //인자로 피자명 또는 피자가게 List와 Random 객체를 입력 받아서 random한 피자명 또는 피자 가게 명을 반환. private String getRandomValueFromList(List list, Random random) { int size = list.size(); int index = random.nextInt(size); return list.get(index); } //random한 피자 메시지를 생성하고, 피자가게 명을 key로 나머지 정보를 value로 하여 Hashmap을 생성하여 반환. public HashMap produce_msg(Faker faker, Random random, int id) { String shopId = getRandomValueFromList(pizzaShop, random); String pizzaName = getRandomValueFromList(pizzaNames, random); String ordId = "ord"+id; String customerName = faker.name().fullName(); String phoneNumber = faker.phoneNumber().phoneNumber(); String address = faker.address().streetAddress(); LocalDateTime now = LocalDateTime.now(); String message = String.format("order_id:%s, shop:%s, pizza_name:%s, customer_name:%s, phone_number:%s, address:%s, time:%s" , ordId, shopId, pizzaName, customerName, phoneNumber, address , now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss", Locale.KOREAN))); //System.out.println(message); HashMap messageMap = new HashMap<>(); messageMap.put("key", shopId); messageMap.put("message", message); return messageMap; } public static void main(String[] args) { PizzaMessage pizzaMessage = new PizzaMessage(); // seed값을 고정하여 Random 객체와 Faker 객체를 생성. long seed = 2022; Random random = new Random(seed); Faker faker = Faker.instance(random); for(int i=0; i < 60; i++) { HashMap message = pizzaMessage.produce_msg(faker, random, i); System.out.println("key:"+ message.get("key") + " message:" + message.get("message")); } } } ``` ```java package org.example.kafka; import com.github.javafaker.Faker; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.HashMap; import java.util.Properties; import java.util.Random; import java.util.concurrent.ExecutionException; import static org.apache.kafka.clients.producer.ProducerConfig.*; public class PizzaProducer { public static final Logger logger = LoggerFactory.getLogger(PizzaProducer.class.getName()); public static void sendPizzaMessage(KafkaProducer kafkaProducer, String topicName, int iterCount, int interIntervalMillis, int intervalMillis, int intervalCount, boolean sync) { PizzaMessage pizzaMessage = new PizzaMessage(); int iterSeq = 0; long seed = 2022; Random random = new Random(seed); Faker faker = Faker.instance(random); while ( iterCount != iterSeq++) { HashMap pMessage = pizzaMessage.produce_msg(faker, random, iterSeq); ProducerRecord producerRecord = new ProducerRecord(topicName, pMessage.get("key"), pMessage.get("message")); sendMessage(kafkaProducer, producerRecord, pMessage, sync); if ((intervalCount > 0) && (iterSeq % intervalCount == 0)) { try { logger.info("############ intervalCount:" + intervalCount + " intervalMillis:" + intervalMillis + " #########"); Thread.sleep(intervalMillis); } catch (InterruptedException e) { logger.error(e.getMessage()); } } if (interIntervalMillis > 0) { try { logger.info("interIntervalMillis:" + interIntervalMillis); Thread.sleep(interIntervalMillis); } catch (InterruptedException e) { logger.error(e.getMessage()); } } } } public static void sendMessage(KafkaProducer kafkaProducer, ProducerRecord producerRecord, HashMap pMessage, boolean sync) { if (!sync) { kafkaProducer.send(producerRecord, (recordMetadata, exception) -> { if (exception == null) { logger.info("async key: {} message:{} partitions: {} offset: {}", pMessage.get("key"), pMessage.get("message"), recordMetadata.partition(), recordMetadata.offset()); } else { logger.error("exception error from broker " + exception.getMessage()); } }); } else { try { RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get(); logger.info("async key: {} message:{} partitions: {} offset: {}", pMessage.get("key"), pMessage.get("message"), recordMetadata.partition(), recordMetadata.offset()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } } public static void main(String[] args) { String topicName = "pizza-topic"; // 3개의 파티션을 가지는 토픽 // KafkaProducer Configuration // null, "hello world" Properties props = new Properties(); // bootstrap.servers, key.serializer.class, value.serializer.class props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // Kafka producer object creation KafkaProducer kafkaProducer = new KafkaProducer(props); sendPizzaMessage(kafkaProducer, topicName, -1, 100, 1000, 100, true); kafkaProducer.close(); } } ``` ### G. producer send acks - `acks`가 0일 때는 producer가 데이터를 보내고 카프카에게 ack를 받지 않고 데이터가 처리되었다고 완료하는 설정 - `acks`가 1일 때는 producer가 카프카에 정상 응답을 받고 데이터 처리를 완료 - `acks`가 all 또는 -1 일 때 producer는 다른 브로커에서 복제해야할 파티션이 모두 복제가 되면 데이터가 정상 처리됨 - `replication-factor`가 3일 때 `min.insync.replicas`가 2 이면 3개의 브로커중 하나가 죽어도 정삭작동한다 - 그러나 3개의 브로커중 2개의 브로커가 죽게 되면 정상 작동을 할 수 없다고 판단하고 `NOT_ENOUGH_REPLICAS`를 에러를 producer에게 전달한다. - `acks`의 default 값은 -1, all 이다 ### H. Record Accumulator, Sender Thread - producer send는 배치형태로 전송이 됩니다. - `linger.ms`는 배치형태로 카프카에게 보내는 시간을 기다리는 시간입니다. (굳이 설정할 필요없음, 20ms 이하로 설정 권장) - `batch.size`는 배치의 사이즈를 정하여 해당 배치의 크기만큼 차게 되면 전송합니다. ### I. producer의 async와 sync 의 배치 - 모두 배치 단위로 동작하지만 동기인 경우 응답이 올 때까지 다음 처리를 하지 않으므로 실제적으로 배치처리가 불가능하다 ### J. delivery.timeout.ms >= linger.ms + request.timeout.ms - 반드시 이렇게 설정되어야 한다. - default 2분으로 설정되어있다. - `request.timeout.ms`의 default는 30초 ### K. max.block.ms - default 60초로 설정되어 있음 - `max.block.ms`는 `Record Accumulator`에 데이터가 꽉 차게 되었을 때 기다리는 시간입니다. ### L. 최대 한번 전송, 적어도 한번 전송, 정확히 한번 전송 - 최대 한번 전송(at most once)은 `acks = 0` - 적어도 한번 전송(at least once)은 `acks = 1 or all` => 중복 허용 - 정확히 한번 전송(exactly once)는 idempotence, 트랜잭션 기반 전송 ### M. idenpotence를 위한 producer 설정 - `enable.idempotence = true` - `acks = all` - `retries > 0` - `max.in.flight.requests.per.connection`은 1~5 사이값 - 기본적으로 적용하는 것을 권장 - 이 옵션을 설정하지 않으면 전송의 순서와 브로커에 저장되는 순서가 변경될 수 있음 - 기본값으로 설정이 되어 있지만 명시적으로 설정해주면 잘못된 설정이 있을 경우 에러가 남 (명시적으로 안해주면 에러 안나고 idenpotence로 동작 안함) ### N. custom partitioner - `defaultPartitioner`의 `stickyPartitionCache`를 사용하지 않고 원하는 키를 파티션 0과 1에만 갈 수 있도록 파티셔닝 구현 - 토픽의 파티션개수는 4개로 정함 ```java package org.example.kafka; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.clients.producer.internals.DefaultPartitioner; import org.apache.kafka.clients.producer.internals.StickyPartitionCache; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.InvalidRecordException; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; import java.util.Map; public class CustomPartitioner implements Partitioner { public static final Logger logger = LoggerFactory.getLogger(CustomPartitioner.class.getName()); private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache(); private String specialKeyName; @Override public void configure(Map configs) { specialKeyName = configs.get("custom.specialKey").toString(); } @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List partitionInfoList = cluster.partitionsForTopic(topic); int numPartitions = partitionInfoList.size(); int numSpecialPartitions = (int)(numPartitions * 0.5); int partitionIndex = 0; if (keyBytes == null) { //return stickyPartitionCache.partition(topic, cluster); throw new InvalidRecordException("key should not be null"); } if (((String) key).equals(specialKeyName)) { partitionIndex = Utils.toPositive(Utils.murmur2(valueBytes)) % numSpecialPartitions; } else { partitionIndex = Utils.toPositive(Utils.murmur2(keyBytes)) % (numPartitions - numSpecialPartitions) + numSpecialPartitions; } logger.info("key: {} is sent to partitions: {}", key.toString(), partitionIndex); return partitionIndex; } @Override public void close() { } } ``` ## (2) Consumer ### Consumer 구조 ``` Fetcher ConsumerNetworkClient SubscriptionState ConsumerCoordinator Heart Beat Thread ``` ### A. Simple Consumer - 1초마다 레코드들을 가져오는 간단한 컨슈머이다 - 지금 구현된 코드는 안전한 컨슈머 종료를 하지 않습니다 - 카프카의 heartbeat에 의해 컨슈머 그룹이 리밸런싱 되었습니다. - 다음 컨슈머에서는 안전한 컨슈머 종료를 시키도록 하겠습니다 - `consumer`는 항상 `group id`를 가지고 있어야 합니다. ```java package org.example.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.List; import java.util.Properties; import static org.apache.kafka.clients.consumer.ConsumerConfig.*; public class SimpleConsumer { public static final Logger logger = LoggerFactory.getLogger(SimpleConsumer.class.getName()); public static void main(String[] args) { String topicName = "simple-topic"; Properties props = new Properties(); props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "ubuntu2:9092"); props.setProperty(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(GROUP_ID_CONFIG, "group-01"); KafkaConsumer kafkaConsumer = new KafkaConsumer(props); kafkaConsumer.subscribe(List.of(topicName)); while (true) { ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : consumerRecords) { logger.info("record => key: {}, value: {}, partition: {}", record.key(), record.value(), record.partition()); } } // kafkaConsumer.close(); } } ``` ### B. consumer fetcher (추후 테스트를 진행해봐야함) - `fetch.min.bytes = 16384`는 카프카 데이터를 읽는 최소의 크기를 말하는 것으로 해당 설정한 바이트보다 커야지 데이터를 즉시 가져온다. - `fetch.max.wait.ms = 500ms`는 카프카에서 데이터가 쌓일 때까지 기다리는 시간이다. 만약 `fetch.min.bytes` 만큼 데이터가 축적되지 않았을 때 기다리는 시간이다. - 데이터가 없어서 기다리는 것이 아니라 데이터가 쌓일 때까지 기다리는 것이다. - `poll`에서 기다리는 시간은 데이터가 없기 때문에 기다리는 것이다. 둘의 정의가 다르다. - `fetch.max.bytes=50MB` - `max.partition.fetch.bytes=1024168`는 오래 전 레코드부터 읽어와야한다면 이 설정이 작동한다. - `max.poll.records=500`은 fetcher가 Liked Queue에서 가져오는 레코드의 개수 ### C. consumer wakeup을 통해 안전하게 종료하기 ```java package org.example.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.List; import java.util.Properties; import static org.apache.kafka.clients.consumer.ConsumerConfig.*; public class ConsumerjWakeup { public static final Logger logger = LoggerFactory.getLogger(ConsumerjWakeup.class.getName()); public static void main(String[] args) { String topicName = "simple-topic"; Properties props = new Properties(); props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "master0:9092"); props.setProperty(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(GROUP_ID_CONFIG, "group-01"); KafkaConsumer kafkaConsumer = new KafkaConsumer(props); kafkaConsumer.subscribe(List.of(topicName)); // main thread Thread mainThread = Thread.currentThread(); // main thread 종료시 별도의 thread로 kafkaconsumer wakeup()메서드를 호출하게 함 Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { logger.info("main program starts to exit by calling wakeup"); kafkaConsumer.wakeup(); try { mainThread.join(); } catch (InterruptedException e) { e.printStackTrace(); } } }); try { while (true) { ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : consumerRecords) { logger.info("record => key: {}, value: {}, partition: {}", record.key(), record.value(), record.partition()); } } } catch (WakeupException e) { logger.error("wakeup exception has been called"); } finally { logger.info("finally consumer is closing"); kafkaConsumer.close(); } } } ``` - `consumer wakeup`을 하기 전에는 `heartbeat expiration`이 발생한다. 그러나 `wakeup`을 한 후에는 카프카 측에서 `group-01`에 있는 컨슈머를 제거한다. ```bash [2022-11-02 09:14:18,106] INFO [GroupCoordinator 0]: Member consumer-group-01-1-2f90ba96-f8e8-4300-838b-ceb931eb2732 in group group-01 has failed, removing it from the group (kafka.coordinator.group.GroupCoordinator) [2022-11-02 09:14:18,106] INFO [GroupCoordinator 0]: Preparing to rebalance group group-01 in state PreparingRebalance with old generation 5 (__consumer_offsets-45) (reason: removing member consumer-group-01-1-2f90ba96-f8e8-4300-838b-ceb931eb2732 on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator) [2022-11-02 09:14:18,107] INFO [GroupCoordinator 0]: Group group-01 with generation 6 is now empty (__consumer_offsets-45) (kafka.coordinator.group.GroupCoordinator) ``` ```bash [2022-11-02 09:08:07,501] INFO [GroupCoordinator 0]: Preparing to rebalance group group-01 in state PreparingRebalance with old generation 3 (__consumer_offsets-45) (reason: Removing member consumer-group-01-1-c6212e3b-5955-44aa-93bd-7d84a23758b1 on LeaveGroup) (kafka.coordinator.group.GroupCoordinator) [2022-11-02 09:08:07,502] INFO [GroupCoordinator 0]: Group group-01 with generation 4 is now empty (__consumer_offsets-45) (kafka.coordinator.group.GroupCoordinator) [2022-11-02 09:08:07,504] INFO [GroupCoordinator 0]: Member MemberMetadata(memberId=consumer-group-01-1-c6212e3b-5955-44aa-93bd-7d84a23758b1, groupInstanceId=None, clientId=consumer-group-01-1, clientHost=/192.168.0.20, sessionTimeoutMs=45000, rebalanceTimeoutMs=300000, supportedProtocols=List(range, cooperative-sticky)) has left group group-01 through explicit `LeaveGroup` request (kafka.coordinator.group.GroupCoordinator) ``` ### D. __consumer_offsets의 정보를 알아보기 1. consumer.config용 config 파일을 생성 ```bash echo "exclude.internal.topics=false" > /kafka/consumer_temp.config ``` 1. __consumer_offsets 토픽을 읽기 ```bash kafka-console-consumer --consumer.config /kafka/consumer_temp.config \ --bootstrap-server localhost:9092 --topic __consumer_offsets \ --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" ``` - `consumer group-01`의 `마지막 offset에서 1을 더한 값`이 `offsetAndMetadata`에 저장이 된다. ```bash [group-01,simple-topic,0]::OffsetAndMetadata(offset=4, leaderEpoch=Optional[0], metadata=, commitTimestamp=1667350382842, expireTimestamp=None) ``` ### E. AUTO_OFFSET_RESET 의 earliest와 latest - `consumer group`이 시작할 때 맨 처음부터 가져올 것인가 최신부터 가져올 것인가를 정하는 옵션이다. - `earliest`를 사용할 때의 주의사항은 `consumer group`에서 처음 `consumer`가 구동될 때는 이것이 작동이 되지만 다운 후 다시 기동할 시에는 맨 처음부터 가져오지 않습니다. - 즉, 이 옵션은 컨슈머 그룹이 시작되고 맨 처음 구동할 때만 작동합니다. ``` # 컨슈머 그룹에서 컨슈머가 작동한 적이 없어서 null [group-01,simple-topic,0]::NULL ``` ``` # 컨슈머 그룹에서 컨슈머가 작동 ( offset은 가져올 offset의 번호이다) [group-01,simple-topic,0]::OffsetAndMetadata(offset=4, leaderEpoch=Optional[0], metadata=, commitTimestamp=1667350230290, expireTimestamp=None) ``` ### F. consumer group status - `empty`는 컨슈머 그룹에 컨슈머가 없을 경우 - `rebalance`는 컨슈머 그룹에 있는 컨슈머들에게 파티션을 할당하는 경우 - `stable`는 파티션 할당이 끝난 후에 안정적인 경우 ``` empty <=> rebalance <=> stable ``` ### G. GROUP_INSTANCE_ID - consumer에게 고정된 그룹 인스턴스 아이디를 부여하여 리밸런싱을 최소화시킨다. - 만약 컨슈머그룹에 3개의 컨슈머가 있게 되고 각각 컨슈머에게 고유한 인스턴스 아이디를 부여하게 되면 컨슈머가 하나 죽더라도 리밸런싱을 하지않는다 - 단, `session.timeout.ms = 45000`에 의해 세션이 끝나면 리밸런싱을 하게 된다 - 그리고 45초 이내에 다시 죽었던 컨슈머가 기동된다면 기존의 할당이 되었던 파티션으로 붙어서 진행이 된다. - 이는 리밸런싱이 일어날 때 컨슈머가 일을 못하는데 일을 계속 처리하면서 리밸런싱이 일어나게 하기 위함이다. ### H. MAX_POLL_INTERVAL_MS_CONFIG - `MAX_POLL_INTERVAL_MS_CONFIG`를 60초로 설정 - 60초가 넘으면 리밸런싱이 일어남 ```java package org.example.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.List; import java.util.Properties; import static org.apache.kafka.clients.consumer.ConsumerConfig.*; public class ConsumerWakeupV2 { public static final Logger logger = LoggerFactory.getLogger(ConsumerWakeupV2.class.getName()); public static void main(String[] args) { String topicName = "pizza-topic-2"; Properties props = new Properties(); props.setProperty(BOOTSTRAP_SERVERS_CONFIG, "master0:9092"); props.setProperty(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(GROUP_ID_CONFIG, "group-02"); props.setProperty(MAX_POLL_INTERVAL_MS_CONFIG, "60000"); KafkaConsumer kafkaConsumer = new KafkaConsumer(props); kafkaConsumer.subscribe(List.of(topicName)); // main thread Thread mainThread = Thread.currentThread(); // main thread 종료시 별도의 thread로 kafkaconsumer wakeup()메서드를 호출하게 함 Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { logger.info("main program starts to exit by calling wakeup"); kafkaConsumer.wakeup(); try { mainThread.join(); } catch (InterruptedException e) { e.printStackTrace(); } } }); int loopCnt = 0; try { while (true) { ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); logger.info("####### loopCnt: {} consumerRecords count: {}", loopCnt++, consumerRecords.count()); for (ConsumerRecord record : consumerRecords) { logger.info("record => key: {}, partition: {}, offset: {}, value: {}", record.key(), record.partition(), record.offset(), record.value()); } try { logger.info("main thread is sleeping {} ms during while loop", loopCnt * 10000); Thread.sleep(loopCnt * 10000); } catch (InterruptedException e) { e.printStackTrace(); } } } catch (WakeupException e) { logger.error("wakeup exception has been called"); } finally { logger.info("finally consumer is closing"); kafkaConsumer.close(); } } } ``` - 처음 조인시에 처음 보내는 poll에서는 데이터를 보내지 않고 루프를 한번 돌린다. ```bash [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Discovered group coordinator master0:9092 (id: 2147483647 rack: null) [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] (Re-)joining group [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Request joining group due to: need to re-join with the given member-id [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] (Re-)joining group [main] INFO org.example.kafka.ConsumerWakeupV2 - ####### loopCnt: 0 consumerRecords count: 0 [main] INFO org.example.kafka.ConsumerWakeupV2 - main thread is sleeping 10000 ms during while loop [main] INFO org.example.kafka.ConsumerWakeupV2 - ####### loopCnt: 1 consumerRecords count: 0 [main] INFO org.example.kafka.ConsumerWakeupV2 - main thread is sleeping 20000 ms during while loop [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Successfully joined group with generation Generation{generationId=2, memberId='consumer-group-02-1-81d557cc-2265-4ff0-9952-b57aad14ce1e', protocol='range'} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Finished assignment for group at generation 2: {consumer-group-02-1-81d557cc-2265-4ff0-9952-b57aad14ce1e=Assignment(partitions=[pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2])} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Successfully synced group in generation Generation{generationId=2, memberId='consumer-group-02-1-81d557cc-2265-4ff0-9952-b57aad14ce1e', protocol='range'} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Notifying assignor about the new Assignment(partitions=[pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2]) [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Adding newly assigned partitions: pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2 [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Setting offset for partition pizza-topic-2-1 to the committed offset FetchPosition{offset=38, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[master0:9092 (id: 0 rack: null)], epoch=0}} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Setting offset for partition pizza-topic-2-2 to the committed offset FetchPosition{offset=31, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[master0:9092 (id: 0 rack: null)], epoch=0}} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Setting offset for partition pizza-topic-2-0 to the committed offset FetchPosition{offset=38, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[master0:9092 (id: 0 rack: null)], epoch=0}} ``` - 로직에서 70초 sleep을 주게 되는데 이때 리밸런싱이 일어난다. ```bash [Consumer clientId=consumer-group-02-1, groupId=group-02] Failing OffsetCommit request since the consumer is not part of an active group [main] WARN org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Synchronous auto-commit of offsets {pizza-topic-2-0=OffsetAndMetadata{offset=171, leaderEpoch=0, metadata=''}, pizza-topic-2-1=OffsetAndMetadata{offset=185, leaderEpoch=0, metadata=''}, pizza-topic-2-2=OffsetAndMetadata{offset=102, leaderEpoch=0, metadata=''}} failed: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group. [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Giving away all assigned partitions as lost since generation has been reset,indicating that consumer is no longer part of the group [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Lost previously assigned partitions pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2 [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] (Re-)joining group [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Request joining group due to: need to re-join with the given member-id [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] (Re-)joining group [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Successfully joined group with generation Generation{generationId=6, memberId='consumer-group-02-1-45d75433-7819-4335-8774-58aeecae1432', protocol='range'} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Finished assignment for group at generation 6: {consumer-group-02-1-45d75433-7819-4335-8774-58aeecae1432=Assignment(partitions=[pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2])} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Successfully synced group in generation Generation{generationId=6, memberId='consumer-group-02-1-45d75433-7819-4335-8774-58aeecae1432', protocol='range'} [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Notifying assignor about the new Assignment(partitions=[pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2]) [main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group-02-1, groupId=group-02] Adding newly assigned partitions: pizza-topic-2-0, pizza-topic-2-1, pizza-topic-2-2 ``` ### I. PARTITION_ASSIGNMENT_STRATEGY_CONFIG - 기본 `Range` 로 설정되어 있으며 멀티 토픽으로 설정되어 있으면 같은 파티션 번호끼리 할당된다. - `Round Robin`는 무작위 파티션번호가 할당된다. - `Range`와 `Round Robin`는 모든 파티션의 소유권을 해제하고 전부 다시 할당하는 것으로 해제하고 할당하는 순간에는 데이터를 처리하지 못한다. - `CooperativeStickyAssignor`는 모든 파티션의 소유권을 해제하는 것이 아닌 기존의 할당된 파티션은 유지한체 다른 컨슈머에게 나눠준다.