카프카 정리와 실습

개념

대용량 실시간 데이터 스트리밍 플랫폼으로, 게시/구독 (pub-sub) 메시징 기능과 분산 로그 저장소의 특징을 모두 갖추고 있다.
높은 처리량과 내결함성을 제공해 대규모 분산 시스템의 데이터 파이프라인에 많이 사용된다.

 

브로커 (Broker)

Kafka 서버 프로세스 1개를 의미한다.
하나의 Kafka 인스턴스가 하나의 브로커로 동작하며, 여러 브로커가 보여 Kafka 클러스터를 구성한다.
일반적으로 프로덕션 환경에서는 3개 이상의 브로커로 클러스터를 구성하여 내결함성 확보를 권장한다.

각 브로커는 자신만의 ID를 가지며 클러스터 내 토픽의 일부 파티션 데이터를 저장하고, 프로듀서/컨슈머로부터 오는 요청을 처리한다.

클러스터에는 하나의 브로커가 "컨트롤러 (Controller) 역할을 맡는다.
컨트롤러 브로커는 클러스터의

• 메타데이터 관리
• 토픽의 파티션 리더 선출
• 브로커 노드의 상태 모니터링
  등의 작업을 수행한다.

만약 브로커에 장애가 발생하면 다른 브로커가 새로운 컨트롤러로 선출되어 클러스터의 운영을 지속한다.

 

토픽과 파티션 (Topic & Partition)

토픽은 Kafka에서 데이터가 분류되어 저장되는 논리적 카테고리를 말한다.
프로듀서는 메시지를 특정 토픽으로 보내고, 컨슈머는 해당 토픽을 구독하여 메시지를 소비한다.

토픽은 로그 파일처럼 동작하여, 입력된 메시지를 순서대로 저장하고 유지한다.
예를 들어 click_log, send_sms 등의 이름으로 토픽을 만들어 그 종류별 데이터를 담을 수 있다.

 

각 토픽은 하나 이상의 파티션(partition) 으로 나뉜다. 파티션은 토픽 내 메시지 로그의 물리적 분할 단위이다.
하나의 토픽에 다수의 파티션이 있으면, 해당 토픽의 메시지들이 여러 브로커에 분산 저장되어 병렬 처리량을 높일 수 있다.
프로듀서는 메시지를 토픽에 보낼 때 내부적으로 특정 파티션에 기록하는데, 별도의 key를 지정하지 않으면 라운드 로빈으로 파티션이 선택되고, 키를 지정하면 키의 해시에 따라 특정 파티션에 매핑된다.

 

컨슈머 측면에서, 하나의 컨슈머 인스턴스는 하나의 파티션만을 독립적으로 읽는다 (컨슈머 그룹 내에서의 제약이다. 같은 그룹 내에서, 하나의 파티션은 하나의 컨슈머 인스턴스에 할당된다.)
따라서 파티션 수를 늘리면 동시에 처리할 수 있는 컨슈머 수를 늘릴 수 있어 병렬 처리 성능이 향상된다.

다만 한번 생성한 파티션 수는 줄일 수 없으므로 초기 설계 시 주의를 기울여야 한다.
(다른 컨슈머 그룹은 동일 파티션을 독립적으로 소비할 수 있다. -> 오프셋을 컨슈머 그룹 단위로 관리하기 때문)

 

또한 파티션은 immutable log구조를 가지므로, 컨슈머가 데이터를 읽어가도 파일에서 삭제되지 않고 그대로 남아있다.
기본 설정에서는 컨슈머가 읽은 여부와 상관없이 일정 기간 동안 데이터가 유지되므로, 새로운 컨슈머 그룹이 같은 토픽을 구독하면 과거 데이터를 다시 처음부터 읽어오는 것도 가능하다. (이 보존 기간은 레코드 보존 정책에 따라 결정된다.)

 

 

리더와 팔로워, ISR (Leader, Follower, In-Sync Replica)

여러 개의 브로커로 이루어진 Kafka 클러스터에서는 데이터의 복제본(replica)을 통해 고가용성을 달성한다.
각 파티션은 클러스터 내 여러 브로커에 복제되어 저장될 수 있는데, 이때 하나의 브로커가 해당 파티션의 리더역할을 맡고, 나머지 복제본은 팔로워로서 리더의 데이터를 실시간 복제한다.

클라이언트(프로듀서와 컨슈머)의 모든 읽기/쓰기 요청은 오직 리더 파티션에게만 전달되며, 팔로워들은 리더의 로그를 따라잡기만 한다.
리더 파티션이 저장한 로그는 팔로워에도 동일한 내용으로 복제되며, 리더 끝부분의 몇 개 레코드를 제외하면 항상 동기화된 동일 사본을 유지한다.

 

Kafka는 토픽 생성 시 복제 계수 (Replication Factor) 를 지정하여, 각 파티션이 몇 개 브로커에 복제될지를 결정한다.
예를 들어, 복제 계수를 3으로 설정하면, 하나의 리더와 두 개의 팔로워 총 3개의 복제본이 유지된다.
이렇게 파티션의 로그를 여러 서버에 복제해 두는 것으로 서버 장애 시 자동으로 다른 복제본을 리더로 승격(fail-over) 함으로써 데이터의 가용성을 높인다.

따라서 한 브로커가 다운되더라도 해당 파티션의 나머지 복제본 중 하나가 새로운 리더가 되어 서비스 중단 없이 메시지를 제공할 수 있다.

복제 시스템에서 중요한 개념으로 ISR이 있다. ISR은 말 그대로 리더와 동기화 상태에 있는 복제본들의 집합을 가리키는데, 해당 파티션의 리더 파티션과 최신 데이터를 갖춘 팔로워 파티션들을 합한 집합이다.

리더는 주기적으로 팔로워들의 상태를 체크하여 지나치게 뒤처지는 팔로워는 ISR에서 제외한다.
프로듀서는 acks설정과 토픽의 min.insync.replicas 설정을 통해 메시지가 ISR에 속한 복제본에 정상 복제되었는지를 기준으로 ack(확인 응답)를 받을 수도 있다.
(예: acks=all일 경우 ISR 내 모든 복제본에 기록될 때까지 대기)

[토픽 파티션의 브로커 간 복제 예시]

Topic1이 3개의 파티션(0, 1, 2)으로 구성된 경우를 보여준다. 세 개 브로커에 파티션들이 분산되어 저장되며,

각 파티션 당 하나의 리더(파란색)와 두 개의 팔로워(흰 색)복제본이 존재한다. 팔로워들은 리더를 따라 로그를 복제하며, 리더+팔로워 복제본 전체가 해당 파티션의 ISR을 이룬다.

 

 

레코드 보존 정책 (Record Retention Policy)

Kafka의 토픽은 기본적으로 영구 저장소가 아니며, 설정된 정책에 따라 오래된 레코드를 자동으로 삭제 또는 압축한다.
레코드 보존 기간/용량을 지정하여 무한정 데이터가 쌓이는 것을 방지할 수 있다.
예를 들어 기본 설정에서는 168시간(7일) 동안 데이터가 보존되고 그 이후에는 가장 오래된 레코드부터 삭제된다.
보존 기간은 브로커의 server.properties의 log.retention.hours설정이나 토픽 생성 시 --config rention-ms=...설정으로 변경할 수 있다.

이외에도

• 용량 기준 보존(log.retention.bytes),
• 로그 세그먼트 단위 삭제주기(log.segment.bytes, logs.segment.ms)
  등 세부 정책을 통해 로그 정리 시점을 정교하게 조정할 수 있다.

 

Kafka는 두 가지 로그 정리 모드를 지원한다.

1. 기본값은 삭제(delete) 모드로, 설정된 기간/용량을 초과하면 가장 오래된 세그먼트 단위로 로그 파일을 삭제한다.
2. 압축(compact) 모드는 키-값 형태의 로그에서 같은 키의 오래된 값들을 정리하고 최근 값만 남겨두는 방식이다. 압축 모드는 주로 상태 저장 토픽이나 상태 변경 이벤트아 같이 현재 상태만 유지하면 되는 데이터흐름에 적합하다. 토픽 생성 시 cleanup.policy=compact 옵션으로 설정할 수 있다. (또는 delete,compact 같이 두 모드 병행도 가능) 

압축의 예: 

(*예: 사용자의 프로필 정보 (user-123 -> name: 홍길동 -> name: 최길동),

CDC(change Data Capture) 시스템에서 변경된 행만 실시간 반영할 때, 

 Kafka Streams의 상태 저장(state store))

 

 

 

정리하면, Kafka는 설정된 보존 기간/용량 내에서는 데이터가 유지되므로,
소비자가 즉시 데이터를 읽지 않더라도 일정 기간 지나지 않는다면 언제든 과거 데이터를 다시 소비할 수 있다.
이는 장애 복구나 새로 합류한 컨슈머 그룹이 초기붙터 재처리를 할 수 있게 해주는 강력한 kafka의 특성이다.

 

 

---

실습

[docker-compose.yml]

version: '3'
services:
  kafka-1:
    image: apache/kafka:latest
    container_name: kafka-1
    hostname: kafka-1
    ports:
      - 19092:9092
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: PLAINTEXT://kafka-1:29092,CONTROLLER://kafka-1:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:29092,PLAINTEXT_HOST://localhost:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  kafka-2:
    image: apache/kafka:latest
    container_name: kafka-2
    hostname: kafka-2
    ports:
      - 19093:9092
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: PLAINTEXT://kafka-2:29092,CONTROLLER://kafka-2:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:29092,PLAINTEXT_HOST://localhost:19093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  kafka-3:
    image: apache/kafka:latest
    container_name: kafka-3
    hostname: kafka-3
    ports:
      - 19094:9092
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: PLAINTEXT://kafka-3:29092,CONTROLLER://kafka-3:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:29092,PLAINTEXT_HOST://localhost:19094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

다중 브로커 클러스터를 구성하면 kafka의 분산성과 내결함성을 확인할 수 있다.

kafka에서는 기본적으로 토픽을 조작하기 위한 CLI도구로 kafka-topics.sh 스크립트를 제공한다.
이를 컨테이너 환경에서 사용하기 위해 컨테이너 내부에서 명령을 실행해야 한다.

docker exec -it -w /opt/kafka/bin kafka-1 sh

Kafka 클라이언트는 부트스트랩 서버 한 곳만 알면 클러스터 정보를 가져와 모든 브로커와 통신할 수 있다 !
(여기서 bootstrap란 "브로커에 처음 연결할 때 사용하는 주소 목록"을 의미하며, 초기 진입점 역할을 한다. )

 

 

---

토픽 생성

[토픽 생성] - (kafka-topics.sh --create)

./kafka-topics.sh --create  --topic test-topic-3  --bootstrap-server kafka-1:29092  --replication-factor 3  --partitions 3

주요 인자로 토픽 이름 (--topic), 파티션 개수 (--partitions), 복제 개수(replication-factor), 그리고 브로커 주소 (--bootstrap-server)를 지정해야 한다.
(나머지 기본값으로는 레코드 보존기간(rention) 기본 7일, 로그 세그먼트 크기 등이 있다)

Created topic test-topic-3.

 

 

 

[토픽 목록 확인]

./kafka-topics.sh --bootstrap-server localhost:29092 --list

현재 클러스터 내에 존재하는 모든 토픽 목록

/opt/kafka/bin $ ./kafka-topics.sh --bootstrap-server kafka-1:29092 --list
test-topic-3
test-topic-3-6

 

 

 

[토픽 상세 정보 조회] - (--describe)

./kafka-topics.sh --bootstrap-server kafka-1:29092 --describe --topic test-topic-3 

특정 토픽의 메타데이터(파티션 구성, 리더/ISR 정보, 설정 등)를 확인

결과

/opt/kafka/bin $ ./kafka-topics.sh --describe --topic test-topic-3 --bootstrap-server kafka-1:29092
Topic: test-topic-3    TopicId: a0dXOEfjSUKFGw4aH7UQIg    PartitionCount: 3    ReplicationFactor: 3    Configs:
    Topic: test-topic-3    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 1    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2    Elr:     LastKnownElr:

 

 

 

[토픽 삭제]

./kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test-topic-3

즉시 토픽이 삭제 (kafka의 삭제는 비동기로 동작하지만, 기본 설정상 바로 토픽을 사용할 수 없도록 표시하고 백그라운드로 삭제를 완료한다.)

참고: Kafka 브로커 설정 중 delete.topic.enable=true일 때만 토픽 삭제가 가능하다. (기본값 true). 만약 false로 설정되어 있으면 --delete 명령을 실행해도 실제로 지워지지 않는다.

 

 

 

---

파티션 추가 및 관리

파티션을 늘리면 데이터의 병렬 처리 가능성이 높아지지만, 이미 존재하는 메시지의 키 분배가 변경되지 않도록 기존 파티션의 데이터는 이동하지 않는다. (증가한 파티션은 새로운 메시지부터 사용 가능)

따라서 파티션 수 증설은 토픽의 정합성이나 순서에 영향을 주지 않지만, 컨슈머 그룹의 rebalancing(재할당)이 발생할 수 있으므로 운영 중에 주의가 필요하다.

[파티션 수 증가] - (--alter --partitions)

./kafka-topics.sh --bootstrap-server kafka-1:29092 --alter --topic test-topic-3 --partitions 6

 

이후 --describe로 확인해보면 PartitionCount가 6으로 증가한 것을 확인할 수 있다.
토픽으로 들어오는 새로운 메시지들은 파티셔닝 로직에 따라 0~5 중 하나의 파티션으로 분배된다.

다중 브로커 클러스터에 파티션을 추가하는 경우, Kafka 컨트롤러가 자동으로 새로운 파티션의 리더/팔로워를 브로커에게 분산할당한다.
필요에 따라 --assignment 옵션으로 직접 어떤 브로커에 파티션을 둘지 지정할 수도 있지만, 일반적으로는 자동할당에 맡긴다.

---

파티션 재할당 및 데이터 이동

만약 브로커를 증설하여 기존 토픽의 파티션을 새 브로커로 재분배(rebalance) 하고 싶다면, Kafka 내장 도구인 kafka-ressign-partitions.sh 스크립트를 사용하면 된다.
JSON으로 토픽 파티션별 원하는 할당을 입력받아 데이터를 이동시켜주는데,
다중 브로커 환경에서 데이터 균형을 맞추거나 특정 브로커의 파티션을 줄이고 싶을 때 사용하는 도구이다.*

---

장애 복구 테스트

3개의 브로커로 클러스터를 구성하고. 복제, 리더 선출, 장애 시 fail-over 등 동작을 확인

[리밸런싱 확인]

test-topic TopicId: kibJcsPSRg-GPkgpUasPcA PartitionCount: 3 ReplicationFactor: 3 Configs: Topic: test-topic Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 Elr: LastKnownElr: Topic: test-topic Partition: 1 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2 Elr: LastKnownElr: Topic: test-topic Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Elr: LastKnownElr:

 

kafka-1, kafka-2, kafka-3 컨테이너 가운데,
docker stop kafka-2로 리더를 중지시킨 뒤

./kafka-topics.sh --describe --topic test-topic-3 --bootstrap-server kafka-1:29092

로 확인해보면

 

<%% 중지 전 %%>
Topic: test-topic-3    TopicId: a0dXOEfjSUKFGw4aH7UQIg    PartitionCount: 3    ReplicationFactor: 3    Configs:
    Topic: test-topic-3    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 1    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2    Elr:     LastKnownElr:

<%% 중지 후 %%>
Topic: test-topic-3    TopicId: a0dXOEfjSUKFGw4aH7UQIg    PartitionCount: 3    ReplicationFactor: 3    Configs:
    Topic: test-topic-3    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,3    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 1    Leader: 3    Replicas: 2,3,1    Isr: 3,1    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1    Elr:     LastKnownElr:

브로커 2가 내려가면, 약간의 지연 후 (수 초 내) Kafka는 해당 브로커가 다운됐음을 감지한다.

partition 1의 리더였던 브로커 2가 사라졌으므로, 브로커 3이 새로운 리더로 승격된다.
브로커가 다운됐음을 감지하면, 컨트롤러는 즉시 ISR에 남아있는 팔로워 중 하나를 새 리더로 선출한다.

 

또 ISR에서도 다운된 브로커 2가 제외되어 각각 팔로워 없이 리더만 ISR로 남은 상태이다.
하지만 kafka는 여전히 가용된다. 브로커 2가 없더라도 남은 브로커들로부터 모든 파티션의 데이터를 서비스할 수 있기 때문이다.

물론 replication-factor 3짜리 토픽들은 일시적으로 복제본이 하나 줄어든 상태라, 추가 장애에는 취약해진다.
브로커 2가 재가동되면 자동으로 해당 파티션들의 팔로워로 합류하여 ISR을 복구한다.

 

 

다시 docker start kafka-2로 컨테이너를 살린 뒤 확인해보면

Topic: test-topic-3    TopicId: a0dXOEfjSUKFGw4aH7UQIg    PartitionCount: 3    ReplicationFactor: 3    Configs:
    Topic: test-topic-3    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,3,2    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 1    Leader: 3    Replicas: 2,3,1    Isr: 3,1,2    Elr:     LastKnownElr:
    Topic: test-topic-3    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2    Elr:     LastKnownElr:

브로커 2가 팔로워로 합류하여 ISR을 복구한 것을 확인할 수 있다.

 

 

---

Producing & Consuming Message

내장 콘솔 도구를 사용해 간단히 동작을 확인해볼 수 있다.

터미널을 2개 열고,

마찬가지로 컨테이너 내부에서 docker exec -it -w /opt/kafka/bin kafka-1 sh
명령을 실행해야 한다.

 

 

[브로커 1 실행] - 컨슈머

./kafka-console-consumer.sh  --topic test-topic-3  --bootstrap-server kafka-1:29092  --group group-rebal


./kafka-console-consumer.sh --topic test-topic-3 --bootstrap-server kafka-1:29092

 

 

[브로커 2 실행] - 컨슈머

./kafka-console-consumer.sh  --topic test-topic-3  --bootstrap-server kafka-2:29092  --group group-rebal


./kafka-console-consumer.sh --topic test-topic-3 --from-beginning --bootstrap-server kafka-3:29092

 

 

[브로커 3 실행] - 프로듀서

./kafka-console-consumer.sh  --topic test-topic-3  --bootstrap-server kafka-2:29092  --group group-rebal


./kafka-console-producer.sh --topic test-topic-3 --bootstrap-server kafka-1:29092

프로듀서에서 입력한 메시지는 브로커 1, 2에서 읽을 수 있다.

아무런 옵션을 주지 않은 브로커 1은 가장 최근 오프셋부터 소비한다.
즉, 토픽의 마지막 오프셋부터 실시간 스트리밍을 한다.
만약 컨슈머를 먼저 실행하고 프로듀서로 메시지를 보내면 새 메시지를 읽을 수 있지만,
컨슈머 실행 전에 있던 메시지는 보지 못한다. (컨슈머의 기본 auto.offset.reset=lastest 설정 때문)

 

 

레프로스펙티브 소비 (--from-beginning)

반대로 브로커 2는 브로커 1과 다르게 --from-beginning 옵션을 추가했는데
이렇게 추가하면 과거에 보내진 메시지를 처음부터 읽게 된다.

해당 컨슈머 그룹은 토픽의 처음(offset=0)부터 데이터를 읽기 시작한다.
이전에 보냈던 메시지들이 순서대로 콘솔에 출력된 뒤, 대기 상태로 다음 메시지를 기다리게 된다.

 

 

정리

컨슈머 그룹은 토픽의 읽기 위치(offset)를 기준으로 동작하며, --from-beginning을 통해 처음부터 읽기(Earliest Offset부터 소비) 가 가능함을 알 수 있다.
이미 과거 메시지를 다 읽은 후에는 실시간 스트림처럼 동작한다.

--from-beginning은 새로운 컨슈머 그룹일 때만 적용되며, 기존에 offset 커밋이 있는 컨슈머 그룹에 적용하려면 auto.offset.reset설정 등이 필요하지만, 콘솔 컨슈머는 매번 랜덤 그룹으로 실행되기 때문에 옵션만 주면 동작한다.

 

콘솔 컨슈머의 기본 동작은 메시지 값을 String으로 출력하는 것이다. 키와 값을 모두 출력하거나 value 외의 key, 파티션, 타임스탬프 등을 함께 확인하고 싶다면
--property print.key=true등의 옵션과 --formatter등을 지정할 수 있다.
예를 들어 --formatter kafka.tools.DefalutMessageFormatter와 적절한 deserializer들을 지정하면 [키]\t[값]형식으로 출력할 수도 있다. 기본값으로는 키는 출력하지 않으며 값만 UTF-8로 표시된다.

---

 

 

 

https://developer.confluent.io/confluent-tutorials/kafka-on-docker/

How to run Kafka locally with Docker