[Kafka/Redis] Kafka 실습

1주차 - Kafka 개념
2주차 - Kafka 설치
3주차 - Kafka와 Spring
4주차 - Kafka 실습
5주차 - Redis개념
6주차 - Kafka/Redis를 활용한 실시간 트랜잭션 집계 서비스 

 

 

1. 이벤트를 쌓기 : 데이터 → kafka → akhq 흐름 확인

 

1) 목표 아키텍처

1. 브라우저(UI): 버튼 클릭/폼 입력 → HTTP 요청 전송
2. 백엔드(API 서버): 요청을 받으면 Kafka Producer로 메시지를 Topic에 발행
3. Kafka Broker: 메시지 저장(로그)
4. AKHQ: Topic에 쌓인 메시지를 UI로 조회(필수)

2) 리파지토리

kafka_study/
  docker-compose.yml
  README.md

  akhq/
    application.yml

  producer-api/
    Dockerfile
    package.json        (또는 build.gradle/pom.xml 등 선택)
    src/...
    .env.example        (선택)

  ui/
    Dockerfile          (선택: nginx로 서빙할 때)
    index.html          (또는 React 등)
    nginx.conf          (선택)

 

3) docker-compose

version: "3.8"

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on:
      - zookeeper
    ports:
      # 호스트(윈도우)에서 Kafka 접근할 때
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"

      # 리스너를 내부/외부로 분리
      KAFKA_LISTENERS: "INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka:29092,EXTERNAL://localhost:9092" ✅
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"

      # 로컬 실습 단순화 옵션
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

  akhq:
    image: tchiotludo/akhq:latest
    depends_on:
      - kafka
    ports:
      - "8080:8080"
    volumes:
      - ./akhq/application.yml:/app/application.yml:ro
    environment:
      # Micronaut이 /app/application.yml을 읽게
      MICRONAUT_CONFIG_FILES: "/app/application.yml"

  producer-api:
    build:
      context: ./producer_api
      dockerfile: Dockerfile
    depends_on:
      - kafka
    environment:
      # Spring에서 이 값을 읽어서 producer 설정에 사용
      KAFKA_BOOTSTRAP_SERVERS: "kafka:29092" ✅
      KAFKA_TOPIC: "demo.ui-events"
      SERVER_PORT: "8082"
    ports:
      - "8082:8082"

  ui:
    build:
      context: ./ui
      dockerfile: Dockerfile
    depends_on:
      - producer-api
    ports:
      - "8081:80"

 

 

▶ kafka bootstrap server 포트와 로컬 포트가 다른 이유 

┌──────────────────────────────────────────────┐
│           Docker 내부 네트워크               │
│                                              │
│  [producer-api]  [akhq]  [zookeeper]         │
│        │           │                         │
│        └─────┬─────┘                         │
│              │                               │
│           [kafka]                            │
│       29092 | 9092                           │
└─────────────┼────────────────────────────────┘
              │  ← 이 구멍만 뚫려있음 (포트 매핑)
         localhost:9092
              │
┌─────────────┼────────────────────────────────┐
│           호스트 (Windows)                   │
│                                              │
│   kafkacat, 브라우저, 기타 프로그램들        │
└──────────────────────────────────────────────┘

• Docker 컨테이너끼리는 서로를 서비스 이름(kafka) 으로 찾을 수 있음
• docker-compose에서 포트 매핑을 선언한 것만 외부로 노출
  • producer-api 컨테이너 입장에서는 kafka:29092로 직접 내부 네트워크를 통해 접근이 가능하기 때문에 굳이 호스트를 거칠 필요 없음
  • 반면 호스트(Windows)에 있는 프로그램은 Docker 내부 네트워크에 접근할 수 없으니, Docker가 뚫어준 구멍인 localhost:9092를 통해서만 들어올 수 있음

▶  리스너를 분리하는 이유

ports:
  - "9092:9092"   # 호스트:컨테이너 → 9092만 뚫음
  # 29092는 매핑 안 함 → 외부에서 접근 불가

그래서 호스트(Windows)에서는:
localhost:9092  → ✅ 접근 가능 (포트 매핑이 있으니까)
localhost:29092 → ❌ 접근 불가 (매핑 안 했으니까)
kafka:29092     → ❌ 접근 불가 (kafka라는 도메인 자체를 모름)

반면 **같은 Docker 네트워크 안의 컨테이너들**은 포트 매핑 없이도 서비스 이름으로 서로 직접 통신할 수 있음
producer-api → kafka:29092  → ✅ (같은 네트워크니까)
producer-api → kafka:9092   → ✅ (이것도 내부에선 가능)

• Kafka는 클라이언트가 처음 연결하면 "앞으로 이 주소로 통신해" 라는 advertised address를 돌려줌
• 호스트에서 localhost:9092로 접속 → Kafka가 "localhost:9092로 통신해" 반환 → ✅ 정상
• producer-api 컨테이너에서 접속 → Kafka가 "localhost:9092로 통신해" 반환 → ❌ 컨테이너 안에서 localhost는 자기 자신이라 연결 실패

성공 화면

 

4) Producer

@PostMapping("/send")
public SendResponse send(@RequestBody SendRequest req) {
    String type = (req.type() == null || req.type().isBlank()) ? "UI_EVENT" : req.type();
    String message = (req.message() == null) ? "" : req.message();

    // JSON 문자열로 보내기 (실무에선 스키마/직렬화로 발전)
    String payload = toJson(type, message);

    kafkaTemplate.send(props.topic(), type, payload);

    return new SendResponse(true, props.topic(), payload);
}

 

▶ topic 파티션 3개 + key별 분배 확인

kafkaTemplate.send(props.topic(), type, payload); 👈 여기서 type 이 key의 역할

VIEW가 왜 CLICK과 같은 partition에 묶이는 걸까

partition = hash(key) % 파티션 수

hash("CLICK") % 3 = 0  → partition 0
hash("VIEW")  % 3 = 0  → partition 0  ← 우연히 같은 파티션!
hash("BUY")   % 3 = 1  → partition 1

"같은 key는 항상 같은 파티션으로 간다"
👉 "다른 key는 다른 파티션으로 간다"는 보장하지 않음

 

 

▶ acks / retries / idempotent producer 

• acks
  ㄴProducer가 “성공”으로 간주하기 전에 브로커로부터 어떤 수준의 확인(ack)을 받을지.
  
  • acks=0 : 안 기다림(가장 빠르지만 유실 가능)
  • acks=1 : 리더 브로커에 기록되면 성공(일반적인 타협)
  • acks=all (-1) : ISR(복제 동기화 집합)까지 확인(가장 안전)
• retries
  ㄴ전송 실패(네트워크 끊김, 리더 선출 등) 시 자동 재시도 횟수.
  • retries가 있으면 “일시적 장애”에 더 강해짐
  • 단, 재시도는 중복 전송 가능성을 만들 수 있음(그래서 idempotence가 중요)
• idempotent producer (enable.idempotence=true)
  ㄴProducer가 재시도하더라도 브로커에 “중복 기록”이 안 되게 막아주는 모드.
  • 켜면 “재시도=중복” 위험이 크게 줄어듦
  • Kafka가 Producer ID / sequence를 써서 중복을 제거함
  • “정확히 한 번(exactly-once)” 전체를 보장하는 건 아니고(소비/처리까지 포함하면 트랜잭션 필요),적어도 ‘브로커에 중복 기록’은 방지하는 쪽이 핵심

spring:
  kafka:
    producer:
      acks: 1 ✅
      retries: 0 ✅ 유실
      properties:
        request.timeout.ms: 30000
        delivery.timeout.ms: 120000
        retry.backoff.ms: 200

 

 

 

5) Consumer Group 만들기 

같은 group.id로 여러 컨슈머를 띄워서, 파티션이 어떻게 나눠 먹는지(리밸런싱), 오프셋 커밋이 어떻게 되는지를 확인

consumer group 생성

cousumer (1) - partition (1) / consumer (2) - partition (0) : 각각의 컨슈머가 각각의 파티션을 담당함

 

UI에서 계속 메시지를 보내면:

• 파티션이 3개라면 보통 3개의 컨슈머가 파티션을 나눠서 메시지를 받게 됨 
• 만약 컨슈머를 4개 띄우면? 남는 1개는 놀게 됨 (파티션 수보다 컨슈머가 많으면 분배 불가)
  • 한 파티션은 같은 그룹 안에서 동시에 두 컨슈머가 소비할 수 없음
  • 즉, “그룹 내 컨슈머 수 ≤ 파티션 수”일 때 효율이 잘 나옴

 

▶ 리밸런싱

consumer (2) 를 종료하자 그때의 partition (1)을 consumer (1)이 읽고 있음

 

 

 

---

2. 데이터 가공 처리 파이프라인 : 쌓인 이벤트를 실시간으로 가공해서 다른 토픽으로 흘려보내는 처리 파이프라인

 

1) (준비) 토픽/컨테이너 구성을 “입력 + 스트림 + 출력” 구조로 확장

• 기존: zookeeper + kafka + akhq + producer-api (+ ui)
• 추가: stream-app(새 Spring Boot 서비스 1개)
• stream-app이 Kafka에 붙을 때는 컨테이너 내부 네트워크 기준으로 kafka:29092 를 바라보는 게 핵심

 

2) 출력 토픽을 하나 더 정하기
ㄴ스트림 처리는 “원본”을 망가뜨리지 않고, 가공 결과를 별도 토픽으로 흘려보내는 게 표준 패턴

• 입력 토픽: 이미 쓰던 demo.ui-events
• 출력 토픽 : demo.stream-events (스트림 결과를 담을 새 토픽)
• 자동 생성 옵션이 켜져 있으면(글에 KAFKA_AUTO_CREATE_TOPICS_ENABLE=true) 스트림 앱이 쓰면서 자동으로 생성

3) Stream 앱(Spring Boot) 프로젝트 만들기

• Spring for Apache Kafka + Kafka Streams
  • “내가 Streams DSL로 토폴로지 짠다(KStream/KTable)” 스타일
  • Spring Kafka가 Kafka Streams를 “1급 지원”해줘서 설정/생명주기 관리가 편해.

4) Streams 설정(application.id, Serde, bootstrap-servers) 잡기

server:
  port: 8085

spring:
  kafka:
	#  앱이 kafka:29092 로 접속 - docker compose 에 29092로 설정
    bootstrap-servers: ${SPRING_KAFKA_BOOTSTRAP_SERVERS:localhost:9092}

    streams:
      application-id: demo-stream-app
      properties:
        # 지금 단계는 문자열로 처리하는 게 제일 단순함
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

        # (선택) 개발 중엔 재시작할 때 상태 꼬임 줄이는 옵션
        # commit.interval.ms: 1000

• application.id (필수): 스트림 앱의 “컨슈머 그룹 + 상태관리” 정체성 (같은 application-id로 여러 인스턴스를 띄우면, Kafka가 파티션을 나눠서 처리)

  • Kafka 브로커 입장에서 보이는 것:
    ┌─────────────────────────────────────┐
    │ Consumer Groups                     │
    │  - demo-stream-app  ← 이게 그거임   │
    │  - my-other-consumer-group          │
    └─────────────────────────────────────┘
    
    Topic: orders (파티션 3개)
    
    demo-stream-app 인스턴스 1개일 때:
      인스턴스A → [파티션0, 파티션1, 파티션2] 전부 담당
    
    demo-stream-app 인스턴스 3개로 스케일아웃:
      인스턴스A → [파티션0]
      인스턴스B → [파티션1]
      인스턴스C → [파티션2]
      
    → 같은 application-id면 Kafka가 자동으로 파티션을 나눠줌

• key/value 직렬화(Serde): 우선은 String으로 시작
• bootstrap servers: 도커 내부니까 kafka:29092 (브로커 1개 존재)

 

5) 토폴로지(처리 로직) 1개 만들기: “변환(map/filter) → 출력 토픽”

╔══════════════════════════════════════════════════════════════════════════╗
║              Kafka Streams — 항상 돌아가는 파이프라인                      ║
╚══════════════════════════════════════════════════════════════════════════╝

  ┌─────────────────────────────────────────────────────────────────────┐
  │  Producer (외부 시스템)                                              │
  │  예) REST API, 프론트엔드 클릭 이벤트                                │
  └───────────────────────────┬─────────────────────────────────────────┘
                              │  발행 (produce)
                              ▼
  ┌───────────────────────────────────────────────────────────────────────┐
  │  📥 SOURCE TOPIC: demo.ui-events                                     │
  │                                                                       │
  │  파티션0: {"type":"CLICK","message":"hi"}                             │
  │  파티션0: {"type":"VIEW","message":"hello"}   ← 이건 필터로 버려짐    │
  │  파티션1: {"type":"CLICK","message":"bye"}                            │
  └───────────────────────────┬───────────────────────────────────────────┘
                              │
                              │  KStream.stream("demo.ui-events")
                              ▼
╔═════════════════════════════════════════════════════════════════════════╗
║                   🔄 KAFKA STREAMS 토폴로지                            ║
║                                                                         ║
║   ┌─────────────────────────────────────────────────────────────────┐   ║
║   │  STEP 1. SOURCE NODE                                            │   ║
║   │  역할: 토픽을 구독해서 레코드를 읽어옴                          │   ║
║   │  코드: builder.stream("demo.ui-events")                         │   ║
║   └───────────────────────────┬─────────────────────────────────────┘   ║
║                               │  레코드 1건씩 흘러내려옴                ║
║                               ▼                                         ║
║   ┌─────────────────────────────────────────────────────────────────┐   ║
║   │  STEP 2. MAP (변환) NODE                                     	│   ║
║   └───────────────────────────┬─────────────────────────────────────┘   ║
║                               │  변환된 레코드                          ║
║                               ▼                                         ║
║   ┌─────────────────────────────────────────────────────────────────┐   ║
║   │  STEP 3. SINK NODE                                              │   ║
║   │  역할: 결과를 출력 토픽에 발행                                  │   ║
║   │  코드: .to("demo.stream-events")                                │   ║
║   └───────────────────────────┬─────────────────────────────────────┘   ║
╚═══════════════════════════════╪═════════════════════════════════════════╝
                                │  발행 (produce)
                                ▼
  ┌─────────────────────────────────────────────────────────────────────┐
  │  📤 SINK TOPIC: demo.stream-events                                  │
  │                                                                     │
  │  파티션0: "CLICK: hi"                                               │
  │  파티션1: "CLICK: bye"                                              │
  └─────────────────────────────────────────────────────────────────────┘
                                │
                                ▼
  ┌─────────────────────────────────────────────────────────────────────┐
  │  Consumer (다운스트림 서비스)                                       │
  │  예) 알림 서버, DB 저장, 대시보드                                   │
  └─────────────────────────────────────────────────────────────────────┘


━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
  핵심 개념
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  ✦ 요청-응답 아님        레코드가 들어오는 순간 자동으로 파이프라인이 실행됨
  ✦ 항상 실행 중          앱이 켜져 있는 한, 토폴로지는 24시간 대기 중
  ✦ 레코드 단위 처리       1건 들어오면 → filter → map → sink 순서로 흐름
  ✦ 그래프 구조           SOURCE → (FILTER) → (MAP) → SINK 노드로 연결된 DAG

• Source: 어떤 토픽에서 읽을지 (builder.stream("demo.ui-events"))
• Processor: 어떻게 가공할지 (filter, mapValues 등)
• Sink: 어디 토픽으로 쓸지 (to("demo.stream-events"))

▶ 중요 개념 : 

• 스트림은 “요청-응답”이 아니라 항상 돌아가는 파이프라인이고,
• Kafka Streams는 내부적으로 “레코드가 들어오면 토폴로지 그래프를 타고 흐른다”는 모델
  • 참고 : 실무에서는 무한 루프를 돌려서 1초에 한번씩 working thread가 polling 하는 방식으로 구현됨 

▶ 결과 : 

ui-event로 producing 하면 topic(stream-event) 적재

• application에서 지정한 consumer group id가 ui-event를 컨슈밍 
• 쉬지 않고 컨슈밍 하여 stream-events topic을 생성함 

 

▶ topic을 여러개 만들어서 보내보기 

@Configuration
@EnableKafkaStreams
public class StreamTopologyConfig {

	private static final String INPUT_TOPIC = "demo.ui-events";

	private static final String CLICK_TOPIC = "demo.click-events";   // A
	private static final String VIEW_TOPIC  = "demo.view-events";    // B
	private static final String UNKNOWN_TOPIC = "demo.unknown-events";

	private final ObjectMapper objectMapper = new ObjectMapper();

	@Bean
	public KStream<String, String> streamPipeline(StreamsBuilder builder) {
		KStream<String, String> source =
			builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String())); ✅ 컨슈밍
            

		// 1) value -> type 추출해서 key를 type으로 맞춰줌(집계/디버깅에 유리)
		KStream<String, String> typed = source
			.filter((k, v) -> v != null && !v.isBlank())
			.selectKey((k, v) -> extractTypeOr(v, "PARSE_ERROR"))
			.peek((k, v) -> System.out.println("[IN ] type=" + k + " value=" + v));

		// 2) branch: 조건에 따라 스트림을 여러 갈래로 분기
		KStream<String, String>[] branches = typed.branch(
			(key, value) -> "CLICK".equalsIgnoreCase(key),
			(key, value) -> "VIEW".equalsIgnoreCase(key),
			(key, value) -> true // 나머지는 전부 여기로
		);

		KStream<String, String> clickStream = branches[0];
		KStream<String, String> viewStream  = branches[1];
		KStream<String, String> unknownStream = branches[2];

		// 3) 각 토픽으로 전송
		clickStream
			.mapValues(v -> "CLICK_STREAM: " + v)
			.peek((k, v) -> System.out.println("[OUT-CLICK] " + k + " " + v))
			.to(CLICK_TOPIC, Produced.with(Serdes.String(), Serdes.String()));

		viewStream
			.mapValues(v -> "VIEW_STREAM: " + v)
			.peek((k, v) -> System.out.println("[OUT-VIEW] " + k + " " + v))
			.to(VIEW_TOPIC, Produced.with(Serdes.String(), Serdes.String()));

		unknownStream
			.mapValues(v -> "UNKNOWN_STREAM: " + v)
			.peek((k, v) -> System.out.println("[OUT-UNKNOWN] " + k + " " + v))
			.to(UNKNOWN_TOPIC, Produced.with(Serdes.String(), Serdes.String()));

		return source;
	}

ui-event 를 컨슈밍 -> 비즈니스 로직 -> topic 발행

 

6) 집계 

 

• 타입별 누적 카운트: CLICK/VIEW가 지금까지 각각 몇 건인지
• 1분 윈도우 카운트: “최근 1분 단위”로 CLICK/VIEW가 몇 건씩 들어왔는지

KStream vs KTable

• KStream: 이벤트의 흐름(append-only). “한 건 들어오면 한 건 처리”
• KTable: “현재 상태(최신 값)” 테이블. 예를 들어 CLICK -> 10 이런 식으로 최신 카운트로 계속 업데이트됨.
• 왜 key가 중요했나? : 집계는 key 단위로 묶어서 계산

 

---

[ELK와 연결해보기] 

 

• 로그를 찍는 주체: producer-api, stream-app
  • Spring Boot, Kafka client, Kafka Streams가 기본적으로 찍는 로그야.
    • 애플리케이션 시작
    • Kafka broker 연결
    • consumer group rebalance
    • Kafka Streams topology 시작
    • 에러 stack trace
• 로그를 읽는 주체: filebeat
• 로그를 가공하는 주체: logstash
• 저장하는 주체: elasticsearch
• 보는 주체: kibana