WebFlux + Kafka + ZooKeeper 완전 정복: 리액티브 스트리밍 아키텍처의 모든 것

         ┌─────────────────────────────────┐
         │         Responsive              │
         │     (빠르고 일관된 응답)           │
         └──────────┬──────────┬───────────┘
                    │          │
         ┌──────────▼──┐  ┌───▼──────────┐
         │  Resilient  │  │   Elastic    │
         │ (장애 복원력) │  │ (부하 적응력)  │
         └──────────┬──┘  └───┬──────────┘
                    │          │
         ┌──────────▼──────────▼───────────┐
         │       Message Driven            │
         │    (비동기 메시지 기반 통신)        │
         └─────────────────────────────────┘

// 명령형: 블로킹 방식
List<User> users = userRepository.findAll();          // DB 호출 - 스레드 블로킹
List<User> activeUsers = new ArrayList<>();
for (User user : users) {
    if (user.isActive()) {
        activeUsers.add(user);
    }
}
List<String> names = new ArrayList<>();
for (User user : activeUsers) {
    names.add(user.getName());
}
return names;

// 리액티브: 논블로킹 방식
return userRepository.findAll()       // Flux<User> 반환 - 논블로킹
    .filter(User::isActive)           // 활성 사용자 필터링
    .map(User::getName)               // 이름 추출
    .collectList();                   // Mono<List<String>> 반환

// 1. Publisher: 데이터를 생산하는 주체
public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

// 2. Subscriber: 데이터를 소비하는 주체
public interface Subscriber<T> {
    void onSubscribe(Subscription s);  // 구독 시작
    void onNext(T t);                  // 데이터 수신
    void onError(Throwable t);         // 에러 발생
    void onComplete();                 // 스트림 완료
}

// 3. Subscription: Publisher-Subscriber 간 연결 관리
public interface Subscription {
    void request(long n);              // n개의 데이터 요청 (백프레셔)
    void cancel();                     // 구독 취소
}

// 4. Processor: Publisher이면서 Subscriber (중간 처리)
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Publisher                     Subscriber
   │                              │
   │◄──── subscribe() ────────────│
   │                              │
   │───── onSubscribe(sub) ──────►│
   │                              │
   │◄──── sub.request(n) ─────────│  ← 백프레셔: n개만 요청
   │                              │
   │───── onNext(data1) ─────────►│
   │───── onNext(data2) ─────────►│
   │───── ...                     │
   │───── onNext(dataN) ─────────►│
   │                              │
   │◄──── sub.request(m) ─────────│  ← 추가 m개 요청
   │                              │
   │───── onComplete() ──────────►│  (또는 onError())

// 1. Buffer: 일정 크기까지 버퍼링 후 오버플로우 시 에러
Flux.range(1, 1000)
    .onBackpressureBuffer(256)
    .subscribe(item -> processSlowly(item));

// 2. Drop: 처리 못하는 데이터는 버림
Flux.range(1, 1000)
    .onBackpressureDrop(dropped ->
        log.warn("Dropped: " + dropped))
    .subscribe(item -> processSlowly(item));

// 3. Latest: 최신 값만 유지
Flux.range(1, 1000)
    .onBackpressureLatest()
    .subscribe(item -> processSlowly(item));

// 4. Error: 처리 못하면 에러 시그널
Flux.range(1, 1000)
    .onBackpressureError()
    .subscribe(item -> processSlowly(item));

Client Request ──► Tomcat Thread Pool (200 threads)
                        │
                   Thread-1 ──► Controller ──► Service ──► DB (블로킹 대기)
                   Thread-2 ──► Controller ──► Service ──► DB (블로킹 대기)
                   Thread-3 ──► Controller ──► Service ──► External API (블로킹 대기)
                   ...
                   Thread-200 ──► (모든 스레드 고갈 시 요청 대기열에 쌓임)

Client Request ──► Netty Event Loop (CPU 코어 수 = N개 스레드)
                        │
                   EventLoop-1 ──► Handler1 ──► (논블로킹 I/O 시작) ──► 다음 요청 처리
                   EventLoop-2 ──► Handler2 ──► (논블로킹 I/O 시작) ──► 다음 요청 처리
                   ...
                   EventLoop-N ──► (I/O 완료 콜백) ──► 응답 전송

┌────────────────────────────────────────────────────┐
│                  Netty Server                      │
│                                                    │
│  ┌──────────────┐    ┌──────────────────────────┐  │
│  │  Boss Group   │    │     Worker Group          │  │
│  │  (1 thread)   │    │  (CPU cores x 2 threads) │  │
│  │              │    │                          │  │
│  │  Accept      │───►│  EventLoop-1: Channel A  │  │
│  │  connections │    │  EventLoop-2: Channel B  │  │
│  │              │    │  EventLoop-3: Channel C  │  │
│  │              │    │  ...                     │  │
│  └──────────────┘    └──────────────────────────┘  │
│                                                    │
│  Channel Pipeline:                                 │
│  ┌─────────┬──────────┬───────────┬────────────┐   │
│  │ Decoder │ Handler1 │ Handler2  │  Encoder   │   │
│  └─────────┴──────────┴───────────┴────────────┘   │
└────────────────────────────────────────────────────┘

// Mono 생성
Mono<String> mono1 = Mono.just("Hello");
Mono<String> mono2 = Mono.empty();
Mono<String> mono3 = Mono.error(new RuntimeException("Error"));

// 실전: 사용자 단건 조회
@GetMapping("/users/me")
public Mono<UserResponse> getCurrentUser(Authentication auth) {
    return userRepository.findByEmail(auth.getName())  // Mono<User>
        .map(UserResponse::from)                        // Mono<UserResponse>
        .switchIfEmpty(Mono.error(new UserNotFoundException()));
}

// Flux 생성
Flux<Integer> flux1 = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> flux2 = Flux.range(1, 100);
Flux<Long> flux3 = Flux.interval(Duration.ofSeconds(1));  // 무한 스트림

// 실전: 사용자 목록 조회
@GetMapping("/users")
public Flux<UserResponse> getActiveUsers() {
    return userRepository.findAll()             // Flux<User>
        .filter(User::isActive)                 // 활성 사용자만
        .map(UserResponse::from)                // DTO 변환
        .take(100);                             // 최대 100명
}

// 1. map: 동기 변환 (1:1)
Flux.just("hello", "world")
    .map(String::toUpperCase)
    // 결과: "HELLO", "WORLD"

// 2. flatMap: 비동기 변환 (1:N, 순서 보장 없음)
Flux.just(1L, 2L, 3L)
    .flatMap(id -> userRepository.findById(id))  // 각 ID로 비동기 조회
    // 결과: 순서가 보장되지 않음 (User2, User1, User3 가능)

// 3. concatMap: 비동기 변환 (1:N, 순서 보장)
Flux.just(1L, 2L, 3L)
    .concatMap(id -> userRepository.findById(id))
    // 결과: 순서 보장 (User1, User2, User3)

// 4. filter: 조건 필터링
Flux.range(1, 10)
    .filter(n -> n % 2 == 0)
    // 결과: 2, 4, 6, 8, 10

// 5. zip: 여러 소스를 조합 (1:1 매칭)
Mono<User> user = userRepository.findById(userId);
Mono<List<Order>> orders = orderRepository.findByUserId(userId).collectList();
Mono.zip(user, orders)
    .map(tuple -> new UserWithOrders(tuple.getT1(), tuple.getT2()));

// 6. merge: 여러 소스를 병합 (인터리빙)
Flux.merge(
    hotNewsFlux,     // 실시간 뉴스
    stockPriceFlux,  // 주식 시세
    alertFlux        // 알림
)
.subscribe(event -> dashboard.update(event));

// 7. concat: 순차적 연결
Flux.concat(
    cacheRepository.findById(id),      // 캐시 먼저
    databaseRepository.findById(id)    // 캐시 미스 시 DB
)
.next();  // 첫 번째 결과만

// 8. switchIfEmpty: 빈 결과 처리
userRepository.findByEmail(email)
    .switchIfEmpty(Mono.defer(() -> createDefaultUser(email)));

// 9. retry + backoff: 재시도 로직
webClient.get()
    .uri("/external-api/data")
    .retrieve()
    .bodyToMono(ExternalData.class)
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .maxBackoff(Duration.ofSeconds(10))
        .filter(ex -> ex instanceof WebClientResponseException.ServiceUnavailable));

// 10. onErrorResume: 에러 복구
userService.getUserProfile(userId)
    .onErrorResume(UserNotFoundException.class,
        ex -> Mono.just(UserProfile.anonymous()))
    .onErrorResume(ServiceException.class,
        ex -> cachedProfileService.getCachedProfile(userId));

// build.gradle
// implementation 'org.springframework.boot:spring-boot-starter-data-r2dbc'
// implementation 'io.r2dbc:r2dbc-postgresql'

// Entity
@Table("users")
public record User(
    @Id Long id,
    String email,
    String name,
    boolean active,
    LocalDateTime createdAt
) {}

// Repository
public interface UserRepository extends ReactiveCrudRepository<User, Long> {

    Flux<User> findByActiveTrue();

    @Query("SELECT * FROM users WHERE email = :email")
    Mono<User> findByEmail(String email);

    @Query("SELECT * FROM users WHERE created_at > :since ORDER BY created_at DESC")
    Flux<User> findRecentUsers(LocalDateTime since);
}

// Service
@Service
@RequiredArgsConstructor
public class UserService {

    private final UserRepository userRepository;
    private final DatabaseClient databaseClient;

    public Flux<User> getActiveUsers() {
        return userRepository.findByActiveTrue();
    }

    // 복잡한 쿼리는 DatabaseClient 사용
    public Flux<UserStats> getUserStats() {
        return databaseClient
            .sql("""
                SELECT u.id, u.name, COUNT(o.id) as order_count
                FROM users u
                LEFT JOIN orders o ON u.id = o.user_id
                GROUP BY u.id, u.name
                HAVING COUNT(o.id) > 0
                """)
            .map(row -> new UserStats(
                row.get("id", Long.class),
                row.get("name", String.class),
                row.get("order_count", Long.class)
            ))
            .all();
    }
}

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient() {
        return WebClient.builder()
            .baseUrl("https://api.example.com")
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .filter(ExchangeFilterFunctions.basicAuthentication("user", "password"))
            .codecs(configurer -> configurer
                .defaultCodecs()
                .maxInMemorySize(16 * 1024 * 1024))  // 16MB
            .build();
    }
}

@Service
@RequiredArgsConstructor
public class ExternalApiService {

    private final WebClient webClient;

    // GET 요청
    public Mono<ProductResponse> getProduct(String productId) {
        return webClient.get()
            .uri("/products/{id}", productId)
            .retrieve()
            .onStatus(HttpStatusCode::is4xxClientError,
                response -> Mono.error(new ProductNotFoundException(productId)))
            .onStatus(HttpStatusCode::is5xxServerError,
                response -> Mono.error(new ExternalServiceException()))
            .bodyToMono(ProductResponse.class)
            .timeout(Duration.ofSeconds(5))
            .retryWhen(Retry.backoff(3, Duration.ofMillis(500)));
    }

    // POST 요청
    public Mono<OrderResponse> createOrder(OrderRequest request) {
        return webClient.post()
            .uri("/orders")
            .bodyValue(request)
            .retrieve()
            .bodyToMono(OrderResponse.class);
    }

    // Streaming 응답
    public Flux<EventData> streamEvents() {
        return webClient.get()
            .uri("/events/stream")
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            .bodyToFlux(EventData.class);
    }
}

┌──────────────────┬────────────┬────────────┬────────────┐
│ 동시접속 수       │ Spring MVC │ WebFlux    │ 차이       │
├──────────────────┼────────────┼────────────┼────────────┤
│ 1,000            │ 8,500 rps  │ 9,200 rps  │ +8%        │
│ 10,000           │ 3,200 rps  │ 9,100 rps  │ +184%      │
│ 100,000          │ 타임아웃    │ 8,800 rps  │ MVC 실패   │
├──────────────────┼────────────┼────────────┼────────────┤
│ 메모리 사용      │ 2.1 GB     │ 0.8 GB     │ -62%       │
│ 스레드 수        │ 204        │ 16         │ -92%       │
│ P99 응답시간     │ 1,200ms    │ 180ms      │ -85%       │
│ (동시접속 10K)   │            │            │            │
└──────────────────┴────────────┴────────────┴────────────┘

// 절대 금지: 이벤트 루프에서 블로킹 호출
@GetMapping("/bad-example")
public Mono<String> badExample() {
    // JDBC는 블로킹 — 이벤트 루프를 멈추게 함
    User user = jdbcTemplate.queryForObject("SELECT * FROM users WHERE id = 1", User.class);
    return Mono.just(user.getName());
}

// 올바른 방법 1: R2DBC 사용
@GetMapping("/good-example-1")
public Mono<String> goodExample1() {
    return r2dbcUserRepository.findById(1L)
        .map(User::getName);
}

// 올바른 방법 2: 블로킹 코드를 별도 스케줄러에서 실행
@GetMapping("/good-example-2")
public Mono<String> goodExample2() {
    return Mono.fromCallable(() ->
            jdbcTemplate.queryForObject("SELECT * FROM users WHERE id = 1", User.class))
        .subscribeOn(Schedulers.boundedElastic())  // 블로킹 전용 스레드풀
        .map(User::getName);
}

// build.gradle
// testImplementation 'io.projectreactor.tools:blockhound:1.0.9.RELEASE'

// 테스트 설정
@BeforeAll
static void setUp() {
    BlockHound.install();
}

@Test
void shouldNotBlock() {
    // 블로킹 코드가 있으면 BlockingOperationError 발생
    StepVerifier.create(userService.getActiveUsers())
        .expectNextCount(5)
        .verifyComplete();
}

┌─────────────────────────────────────────────────────────────┐
│                     Kafka Cluster                           │
│                                                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │  Broker 1   │  │  Broker 2   │  │  Broker 3   │         │
│  │             │  │             │  │             │         │
│  │ Topic-A P0  │  │ Topic-A P1  │  │ Topic-A P2  │  Leader │
│  │ Topic-B P1R │  │ Topic-B P0  │  │ Topic-A P0R │  Replica│
│  │ Topic-A P2R │  │ Topic-A P0R │  │ Topic-B P1  │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
│                                                             │
│  P = Partition (Leader),  R = Replica (Follower)            │
└─────────────────────────────────────────────────────────────┘
         ▲                                    │
         │                                    ▼
   ┌─────────────┐                   ┌─────────────────┐
   │  Producers  │                   │  Consumer Group  │
   │             │                   │                 │
   │ App1, App2  │                   │ Consumer1 ← P0  │
   │             │                   │ Consumer2 ← P1  │
   │             │                   │ Consumer3 ← P2  │
   └─────────────┘                   └─────────────────┘

// Producer 설정
@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        // 성능 튜닝
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);        // 32KB 배치
        config.put(ProducerConfig.LINGER_MS_CONFIG, 20);            // 20ms 대기
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");  // 압축
        config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);  // 64MB 버퍼

        // 안정성
        config.put(ProducerConfig.ACKS_CONFIG, "all");              // 모든 ISR 확인
        config.put(ProducerConfig.RETRIES_CONFIG, 3);               // 재시도
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 멱등성 보장

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

// Consumer 설정
@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service-group");

        // 오프셋 관리
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);  // 수동 커밋

        // 성능 튜닝
        config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);      // 1KB 최소 패치
        config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);     // 500ms 대기
        config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);      // 최대 500건

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);  // 3개의 컨슈머 스레드
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

// Consumer 리스너
@Service
@Slf4j
public class OrderEventConsumer {

    @KafkaListener(
        topics = "order-events",
        groupId = "order-service-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            ConsumerRecord<String, OrderEvent> record,
            Acknowledgment ack) {
        try {
            log.info("Received: topic={}, partition={}, offset={}, key={}",
                record.topic(), record.partition(), record.offset(), record.key());

            OrderEvent event = record.value();
            processOrderEvent(event);

            ack.acknowledge();  // 수동 커밋
        } catch (Exception e) {
            log.error("Failed to process order event", e);
            // 에러 처리: DLQ(Dead Letter Queue)로 전송하거나 재시도
        }
    }
}

Topic: order-events (6 partitions)

Consumer Group: order-service-group
  Consumer-1: P0, P1  (2개 파티션 담당)
  Consumer-2: P2, P3  (2개 파티션 담당)
  Consumer-3: P4, P5  (2개 파티션 담당)

규칙: 하나의 파티션은 그룹 내 하나의 컨슈머만 소비 가능
     컨슈머 수 > 파티션 수이면 일부 컨슈머는 유휴 상태

// Transactional Producer 설정
config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-producer-1");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// 트랜잭션 사용 예제
@Service
@RequiredArgsConstructor
public class OrderProcessor {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void processOrder(Order order) {
        kafkaTemplate.executeInTransaction(ops -> {
            // 하나의 트랜잭션으로 묶임
            ops.send("order-confirmed", order.getId(), new OrderConfirmedEvent(order));
            ops.send("inventory-update", order.getProductId(), new InventoryUpdateEvent(order));
            ops.send("notification", order.getUserId(), new NotificationEvent(order));
            return true;
        });
        // 3개의 메시지가 모두 성공하거나 모두 실패
    }
}

// KStream: 이벤트 스트림 (각 레코드가 독립적)
// KTable: 변경 로그 (키의 최신 값만 유지)

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean
    public KStream<String, OrderEvent> orderStream(StreamsBuilder builder) {
        KStream<String, OrderEvent> stream = builder.stream("order-events");

        // 1. 필터링 + 변환
        stream
            .filter((key, event) -> event.getAmount() > 10000)  // 1만원 이상
            .mapValues(event -> new HighValueOrderEvent(event))
            .to("high-value-orders");

        // 2. 윈도우 집계: 5분 간격 주문 금액 합계
        stream
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .aggregate(
                () -> 0L,
                (key, event, total) -> total + event.getAmount(),
                Materialized.as("order-amount-store")
            )
            .toStream()
            .to("order-amount-per-5min");

        // 3. KStream-KTable Join
        KTable<String, UserProfile> userTable = builder.table("user-profiles");
        stream
            .join(userTable,
                (order, user) -> new EnrichedOrder(order, user))
            .to("enriched-orders");

        return stream;
    }
}

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-server",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "password",
    "database.server.id": "1",
    "topic.prefix": "myapp",
    "database.include.list": "orders_db",
    "table.include.list": "orders_db.orders,orders_db.order_items",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes"
  }
}

// user-event.avsc (Avro 스키마)
{
  "type": "record",
  "name": "UserEvent",
  "namespace": "com.example.events",
  "fields": [
    {"name": "userId", "type": "string"},
    {"name": "action", "type": {"type": "enum", "name": "Action",
      "symbols": ["CREATED", "UPDATED", "DELETED"]}},
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}

┌──────────────────────────────────────────────────────────────┐
│                    Producer 튜닝                              │
├──────────────────┬───────────────────────────────────────────┤
│ batch.size       │ 16KB(기본) → 32~64KB (처리량 증가)         │
│ linger.ms        │ 0(기본) → 10~50ms (배치 효율성)            │
│ compression.type │ none → lz4 (속도), zstd (압축률)           │
│ buffer.memory    │ 32MB(기본) → 64~128MB (대량 전송 시)       │
│ max.in.flight    │ 5(기본) → 1 (순서 보장 필요 시)            │
├──────────────────┴───────────────────────────────────────────┤
│                    Consumer 튜닝                              │
├──────────────────┬───────────────────────────────────────────┤
│ fetch.min.bytes  │ 1(기본) → 1024~10240 (배치 페치)           │
│ fetch.max.wait   │ 500ms(기본) → 조정 (지연 vs 처리량)        │
│ max.poll.records │ 500(기본) → 워크로드에 따라 조정            │
│ session.timeout  │ 45s(기본) → 리밸런싱 민감도 조절            │
├──────────────────┴───────────────────────────────────────────┤
│                    Broker 튜닝                                │
├──────────────────┬───────────────────────────────────────────┤
│ num.partitions   │ 토픽 기본 파티션 수 (목표 처리량 / 파티션당)│
│ replication      │ 3 (프로덕션 권장)                          │
│ min.insync       │ 2 (acks=all과 함께 사용)                   │
│ log.retention    │ 7일(기본) → 비즈니스 요구에 따라             │
└──────────────────┴───────────────────────────────────────────┘

목표 처리량 = T (messages/sec)
단일 파티션 처리량 = P (messages/sec)
필요 파티션 수 = max(T/P_producer, T/P_consumer)

예시:
- 목표: 100,000 msg/sec
- Producer 단일 파티션: 50,000 msg/sec
- Consumer 단일 파티션: 10,000 msg/sec
- 필요 파티션: max(100K/50K, 100K/10K) = max(2, 10) = 10개

실무 권장:
- 소규모: 6~12 파티션
- 중규모: 12~30 파티션
- 대규모: 30~100 파티션
- 초대규모: 100+ (KRaft 모드에서 100만개 이상 지원)

┌────────────────────────────────────────────┐
│              ZooKeeper Ensemble             │
│                                            │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│  │  ZK Node  │ │  ZK Node  │ │  ZK Node  │   │
│  │  (Leader) │ │(Follower)│ │(Follower)│   │
│  └──────────┘ └──────────┘ └──────────┘   │
│                                            │
│  저장 데이터:                                │
│  /brokers/ids/[0,1,2]     - 브로커 목록     │
│  /brokers/topics/my-topic - 토픽 메타데이터  │
│  /controller              - 컨트롤러 선출    │
│  /admin/delete_topics     - 토픽 삭제 요청   │
│  /config                  - 동적 설정        │
└────────────────────────────────────────────┘
           ▲          ▲          ▲
           │          │          │
     Kafka Broker1  Broker2   Broker3

┌────────────────────────────────────────────────────┐
│              Kafka Cluster (KRaft Mode)             │
│                                                    │
│  ┌───────────────────────────────────────────┐     │
│  │         Controller Quorum                  │     │
│  │                                           │     │
│  │  Broker-0      Broker-1      Broker-2     │     │
│  │  (Controller   (Controller   (Controller  │     │
│  │   + Broker)     + Broker)     + Broker)   │     │
│  │  [Active]      [Follower]    [Follower]   │     │
│  └───────────────────────────────────────────┘     │
│                                                    │
│  ┌───────────────────────────────────────────┐     │
│  │         Data Brokers                       │     │
│  │  Broker-3      Broker-4      Broker-5     │     │
│  │  (Broker only) (Broker only) (Broker only)│     │
│  └───────────────────────────────────────────┘     │
│                                                    │
│  Metadata Log: __cluster_metadata (내부 토픽)       │
│  합의 프로토콜: Raft (자체 구현)                      │
└────────────────────────────────────────────────────┘

# 1단계: 클러스터 ID 확인
bin/kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/00000000000000000000.log --cluster-id

# 2단계: ZK에서 메타데이터 마이그레이션 준비
bin/kafka-storage.sh format \
  --config config/kraft/server.properties \
  --cluster-id YOUR_CLUSTER_ID \
  --release-version 3.7

# 3단계: 브로커 설정 변경 (server.properties)
# ZK 모드 설정 제거
# zookeeper.connect=localhost:2181  ← 제거

# KRaft 모드 설정 추가
# process.roles=broker,controller
# node.id=1
# controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
# controller.listener.names=CONTROLLER
# listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# 4단계: 롤링 재시작 (하나씩 순차적으로)
# 각 브로커를 중지 → KRaft 설정 적용 → 재시작

# 5단계: 마이그레이션 완료 확인
bin/kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/00000000000000000000.log --verify

@RestController
@RequestMapping("/api/stream")
public class SseController {

    private final StockPriceService stockPriceService;

    // 방법 1: Flux + ServerSentEvent 래퍼
    @GetMapping(value = "/stocks", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<StockPrice>> streamStockPrices(
            @RequestParam List<String> symbols) {
        return stockPriceService.getPriceStream(symbols)
            .map(price -> ServerSentEvent.<StockPrice>builder()
                .id(String.valueOf(price.getTimestamp()))
                .event("price-update")
                .data(price)
                .retry(Duration.ofSeconds(5))
                .build());
    }

    // 방법 2: 심플한 Flux 스트림
    @GetMapping(value = "/notifications", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<NotificationEvent> streamNotifications(
            @AuthenticationPrincipal UserDetails user) {
        return notificationService.getNotificationStream(user.getUsername())
            .doOnCancel(() -> log.info("Client disconnected: {}", user.getUsername()));
    }

    // 방법 3: Heartbeat 포함 (연결 유지)
    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Object>> streamWithHeartbeat() {
        Flux<ServerSentEvent<Object>> dataStream = eventService.getEventStream()
            .map(event -> ServerSentEvent.builder()
                .event("data")
                .data((Object) event)
                .build());

        Flux<ServerSentEvent<Object>> heartbeat = Flux.interval(Duration.ofSeconds(30))
            .map(tick -> ServerSentEvent.builder()
                .event("heartbeat")
                .data((Object) "ping")
                .build());

        return Flux.merge(dataStream, heartbeat);
    }
}

const eventSource = new EventSource('/api/stream/stocks?symbols=AAPL,GOOGL,MSFT')

eventSource.addEventListener('price-update', (event) => {
  const price = JSON.parse(event.data)
  updateDashboard(price)
})

eventSource.addEventListener('heartbeat', () => {
  console.log('Connection alive')
})

eventSource.onerror = (error) => {
  console.error('SSE error:', error)
  // EventSource는 자동으로 재연결을 시도합니다
}

@Configuration
public class WebSocketConfig {

    @Bean
    public HandlerMapping webSocketHandlerMapping(ChatWebSocketHandler handler) {
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/ws/chat", handler);

        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
        mapping.setUrlMap(map);
        mapping.setOrder(-1);
        return mapping;
    }

    @Bean
    public WebSocketHandlerAdapter handlerAdapter() {
        return new WebSocketHandlerAdapter();
    }
}

@Component
@Slf4j
public class ChatWebSocketHandler implements WebSocketHandler {

    private final Sinks.Many<ChatMessage> chatSink =
        Sinks.many().multicast().onBackpressureBuffer();

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // 인바운드: 클라이언트 → 서버
        Mono<Void> input = session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .map(this::parseChatMessage)
            .doOnNext(msg -> {
                log.info("Received: {}", msg);
                chatSink.tryEmitNext(msg);  // 모든 구독자에게 브로드캐스트
            })
            .then();

        // 아웃바운드: 서버 → 클라이언트
        Mono<Void> output = session.send(
            chatSink.asFlux()
                .map(msg -> session.textMessage(toJson(msg)))
        );

        // 양방향 동시 처리
        return Mono.zip(input, output).then();
    }

    private ChatMessage parseChatMessage(String payload) {
        return new ObjectMapper().readValue(payload, ChatMessage.class);
    }

    private String toJson(ChatMessage msg) {
        return new ObjectMapper().writeValueAsString(msg);
    }
}

syntax = "proto3";

package stockservice;

service StockService {
  // Server Streaming: 서버가 연속으로 전송
  rpc StreamPrices(StockRequest) returns (stream StockPrice);

  // Client Streaming: 클라이언트가 연속으로 전송
  rpc SendOrders(stream OrderRequest) returns (OrderSummary);

  // Bidirectional Streaming: 양방향
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message StockRequest {
  repeated string symbols = 1;
}

message StockPrice {
  string symbol = 1;
  double price = 2;
  int64 timestamp = 3;
}

message OrderRequest {
  string symbol = 1;
  int32 quantity = 2;
  double price = 3;
}

message OrderSummary {
  int32 total_orders = 1;
  double total_amount = 2;
}

message ChatMessage {
  string user = 1;
  string content = 2;
  int64 timestamp = 3;
}

@GrpcService
public class StockGrpcService extends StockServiceGrpc.StockServiceImplBase {

    private final StockPriceService priceService;

    @Override
    public void streamPrices(StockRequest request,
                             StreamObserver<StockPrice> responseObserver) {
        // 구독 시작
        Disposable subscription = priceService
            .getPriceStream(request.getSymbolsList())
            .subscribe(
                price -> {
                    // 각 가격 업데이트를 클라이언트에 전송
                    responseObserver.onNext(StockPrice.newBuilder()
                        .setSymbol(price.getSymbol())
                        .setPrice(price.getPrice())
                        .setTimestamp(price.getTimestamp())
                        .build());
                },
                error -> {
                    responseObserver.onError(Status.INTERNAL
                        .withDescription(error.getMessage())
                        .asRuntimeException());
                },
                () -> {
                    responseObserver.onCompleted();
                }
            );

        // 클라이언트 연결 해제 시 구독 취소
        Context.current().addListener(
            context -> subscription.dispose(),
            Runnable::run
        );
    }
}

                    ┌─────────────┐
                    │ 양방향 통신  │
                    │  필요한가?   │
                    └──────┬──────┘
                     예 /     \ 아니오
                      /       \
            ┌────────▼──┐  ┌──▼──────────┐
            │ 브라우저    │  │    SSE      │
            │ 직접 통신?  │  │ (가장 간단)  │
            └─────┬─────┘  └─────────────┘
              예 /   \ 아니오
               /     \
    ┌─────────▼─┐  ┌─▼──────────────┐
    │ WebSocket │  │ gRPC Streaming │
    │ (브라우저)  │  │ (MSA/백엔드)    │
    └───────────┘  └────────────────┘

// build.gradle
// implementation 'io.projectreactor.kafka:reactor-kafka:1.3.23'

@Configuration
public class ReactorKafkaConfig {

    @Bean
    public ReactiveKafkaProducerTemplate<String, Object> reactiveProducer() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        SenderOptions<String, Object> senderOptions = SenderOptions.create(props);
        return new ReactiveKafkaProducerTemplate<>(senderOptions);
    }

    @Bean
    public ReactiveKafkaConsumerTemplate<String, StockPrice> reactiveConsumer() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "stock-price-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        ReceiverOptions<String, StockPrice> receiverOptions =
            ReceiverOptions.<String, StockPrice>create(props)
                .subscription(Collections.singleton("stock-prices"));

        return new ReactiveKafkaConsumerTemplate<>(receiverOptions);
    }
}

┌──────────┐    ┌─────────┐    ┌───────────┐    ┌──────────┐    ┌──────────┐
│ External │    │  Kafka  │    │  WebFlux  │    │   SSE    │    │  React   │
│   API    │───►│ Producer│───►│ Consumer  │───►│ Endpoint │───►│ Frontend │
│          │    │         │    │ (Reactor) │    │          │    │          │
└──────────┘    └─────────┘    └───────────┘    └──────────┘    └──────────┘
  시세 소스        토픽 발행       리액티브 소비     HTTP 스트림     실시간 렌더링

// 1. Kafka Producer: 외부 API에서 시세 데이터를 Kafka로
@Service
@RequiredArgsConstructor
@Slf4j
public class StockPriceProducer {

    private final ReactiveKafkaProducerTemplate<String, Object> producer;
    private final WebClient marketDataClient;

    @Scheduled(fixedRate = 1000)  // 1초마다
    public void fetchAndPublish() {
        marketDataClient.get()
            .uri("/v1/quotes?symbols=AAPL,GOOGL,MSFT")
            .retrieve()
            .bodyToFlux(StockPrice.class)
            .flatMap(price -> producer.send("stock-prices", price.getSymbol(), price))
            .doOnNext(result -> log.debug("Sent: topic={}, partition={}, offset={}",
                result.recordMetadata().topic(),
                result.recordMetadata().partition(),
                result.recordMetadata().offset()))
            .subscribe();
    }
}

// 2. Kafka Consumer + SSE: Kafka에서 브라우저로
@RestController
@RequestMapping("/api/stream")
@RequiredArgsConstructor
public class StockStreamController {

    private final ReactiveKafkaConsumerTemplate<String, StockPrice> consumer;

    // Kafka 토픽을 SSE로 브리지
    @GetMapping(value = "/stock-prices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<StockPrice>> streamStockPrices(
            @RequestParam(required = false) List<String> symbols) {

        return consumer.receiveAutoAck()
            .map(ConsumerRecord::value)
            .filter(price -> symbols == null || symbols.contains(price.getSymbol()))
            .map(price -> ServerSentEvent.<StockPrice>builder()
                .id(price.getSymbol() + "-" + price.getTimestamp())
                .event("price-update")
                .data(price)
                .build())
            .doOnCancel(() -> log.info("Client disconnected from stock stream"));
    }
}

┌──────────────────────────────────────────────────────────┐
│                    CQRS Pattern                          │
│                                                          │
│  Command Side              │  Query Side                 │
│                            │                             │
│  ┌──────────┐              │  ┌───────────┐              │
│  │ WebFlux  │  Command     │  │  WebFlux  │  Query       │
│  │Controller│────────┐     │  │Controller │◄─────┐       │
│  └──────────┘        │     │  └───────────┘      │       │
│                      ▼     │                     │       │
│  ┌──────────────────────┐  │  ┌─────────────────────┐   │
│  │   Kafka Topic        │  │  │   Read Database      │   │
│  │   (Event Store)      │──┼─►│   (Materialized     │   │
│  │   order-events       │  │  │    View / Redis)     │   │
│  └──────────────────────┘  │  └─────────────────────┘   │
│                            │                             │
└──────────────────────────────────────────────────────────┘

// Command: 주문 생성 → Kafka에 이벤트 발행
@Service
@RequiredArgsConstructor
public class OrderCommandService {

    private final ReactiveKafkaProducerTemplate<String, Object> producer;

    public Mono<String> createOrder(CreateOrderCommand command) {
        String orderId = UUID.randomUUID().toString();
        OrderCreatedEvent event = new OrderCreatedEvent(
            orderId, command.getUserId(), command.getItems(),
            command.getTotalAmount(), Instant.now()
        );

        return producer.send("order-events", orderId, event)
            .map(result -> orderId);
    }
}

// Query: Kafka 이벤트를 소비하여 읽기 모델 업데이트
@Service
@RequiredArgsConstructor
public class OrderProjectionService {

    private final ReactiveKafkaConsumerTemplate<String, OrderEvent> consumer;
    private final ReactiveRedisTemplate<String, OrderView> redisTemplate;

    @PostConstruct
    public void startProjection() {
        consumer.receiveAutoAck()
            .map(ConsumerRecord::value)
            .flatMap(this::updateProjection)
            .doOnError(e -> log.error("Projection error", e))
            .retry()
            .subscribe();
    }

    private Mono<Boolean> updateProjection(OrderEvent event) {
        if (event instanceof OrderCreatedEvent created) {
            OrderView view = OrderView.from(created);
            return redisTemplate.opsForValue()
                .set("order:" + created.getOrderId(), view);
        }
        return Mono.just(true);
    }
}

// Query Controller: Redis에서 읽기
@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderQueryController {

    private final ReactiveRedisTemplate<String, OrderView> redisTemplate;

    @GetMapping("/{orderId}")
    public Mono<OrderView> getOrder(@PathVariable String orderId) {
        return redisTemplate.opsForValue()
            .get("order:" + orderId)
            .switchIfEmpty(Mono.error(new OrderNotFoundException(orderId)));
    }
}

┌────────────────────────────────────────────────────────────────────┐
│                    Real-Time Stock Price System                    │
│                                                                    │
│  ┌──────────┐   ┌──────────┐   ┌────────────┐   ┌──────────────┐ │
│  │ Market   │   │  Kafka   │   │  WebFlux   │   │   React      │ │
│  │ Data API │──►│ Producer │──►│  Consumer  │──►│  Frontend    │ │
│  │ (Yahoo/  │   │          │   │  + SSE     │   │  (Dashboard) │ │
│  │  Alpha)  │   │          │   │            │   │              │ │
│  └──────────┘   └──────────┘   └────────────┘   └──────────────┘ │
│       │              │               │                  │         │
│       │         ┌────▼────┐    ┌─────▼─────┐    ┌──────▼──────┐ │
│       │         │  Kafka  │    │  Redis     │    │ Prometheus  │ │
│       │         │  Topic  │    │  Cache     │    │ + Grafana   │ │
│       │         │ (3 rep) │    │            │    │             │ │
│       │         └─────────┘    └───────────┘    └─────────────┘ │
│       │                                                          │
│  ┌────▼──────────────────────────────────────┐                   │
│  │  Kafka Streams: 집계/분석                   │                   │
│  │  - 5분 이동평균                              │                   │
│  │  - 일일 최고/최저가                          │                   │
│  │  - 거래량 이상 탐지                          │                   │
│  └───────────────────────────────────────────┘                   │
└────────────────────────────────────────────────────────────────────┘

@Component
@RequiredArgsConstructor
@Slf4j
public class MarketDataFetcher {

    private final WebClient marketClient;
    private final ReactiveKafkaProducerTemplate<String, Object> kafkaProducer;

    @Scheduled(fixedRate = 1000)
    public void fetchPrices() {
        List<String> symbols = List.of("AAPL", "GOOGL", "MSFT", "AMZN", "TSLA");

        Flux.fromIterable(symbols)
            .flatMap(symbol -> marketClient.get()
                .uri("/quote/{symbol}", symbol)
                .retrieve()
                .bodyToMono(MarketQuote.class)
                .map(quote -> new StockPrice(
                    symbol, quote.getPrice(), quote.getVolume(),
                    quote.getChange(), Instant.now().toEpochMilli()))
                .onErrorResume(e -> {
                    log.warn("Failed to fetch {}: {}", symbol, e.getMessage());
                    return Mono.empty();
                }))
            .flatMap(price -> kafkaProducer
                .send("stock-prices", price.getSymbol(), price)
                .doOnSuccess(r -> log.debug("Published: {} = {}",
                    price.getSymbol(), price.getPrice())))
            .subscribe();
    }
}

@Configuration
@EnableKafkaStreams
public class StockAnalyticsConfig {

    @Bean
    public KStream<String, StockPrice> stockAnalytics(StreamsBuilder builder) {
        KStream<String, StockPrice> priceStream = builder.stream("stock-prices");

        // 5분 이동평균 계산
        priceStream
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .aggregate(
                PriceAggregation::new,
                (symbol, price, agg) -> agg.add(price.getPrice()),
                Materialized.<String, PriceAggregation, WindowStore<Bytes, byte[]>>
                    as("price-5min-avg")
                    .withValueSerde(new PriceAggregationSerde()))
            .toStream()
            .map((windowed, agg) -> KeyValue.pair(
                windowed.key(),
                new PriceAverage(windowed.key(), agg.getAverage(),
                    agg.getMin(), agg.getMax(), agg.getCount())))
            .to("stock-price-analytics");

        return priceStream;
    }
}

// 집계 클래스
public class PriceAggregation {
    private double sum = 0;
    private double min = Double.MAX_VALUE;
    private double max = Double.MIN_VALUE;
    private long count = 0;

    public PriceAggregation add(double price) {
        sum += price;
        min = Math.min(min, price);
        max = Math.max(max, price);
        count++;
        return this;
    }

    public double getAverage() {
        return count > 0 ? sum / count : 0;
    }

    // getters ...
}

@RestController
@RequestMapping("/api/stocks")
@RequiredArgsConstructor
public class StockStreamController {

    private final ReactiveKafkaConsumerTemplate<String, StockPrice> priceConsumer;
    private final ReactiveKafkaConsumerTemplate<String, PriceAverage> analyticsConsumer;

    // 실시간 시세 스트림
    @GetMapping(value = "/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<StockPrice>> liveStream(
            @RequestParam List<String> symbols) {
        return priceConsumer.receiveAutoAck()
            .map(ConsumerRecord::value)
            .filter(p -> symbols.contains(p.getSymbol()))
            .map(p -> ServerSentEvent.<StockPrice>builder()
                .event("price")
                .data(p)
                .build());
    }

    // 분석 데이터 스트림
    @GetMapping(value = "/analytics", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<PriceAverage>> analyticsStream() {
        return analyticsConsumer.receiveAutoAck()
            .map(ConsumerRecord::value)
            .map(avg -> ServerSentEvent.<PriceAverage>builder()
                .event("analytics")
                .data(avg)
                .build());
    }
}

// hooks/useStockStream.ts
import { useState, useEffect, useCallback } from 'react';

interface StockPrice {
  symbol: string;
  price: number;
  volume: number;
  change: number;
  timestamp: number;
}

export function useStockStream(symbols: string[]) {
  const [prices, setPrices] = useState<Map<string, StockPrice>>(new Map());

  useEffect(() => {
    const params = symbols.map(s => `symbols=${s}`).join('&');
    const eventSource = new EventSource(`/api/stocks/live?${params}`);

    eventSource.addEventListener('price', (event: MessageEvent) => {
      const price: StockPrice = JSON.parse(event.data);
      setPrices(prev => new Map(prev).set(price.symbol, price));
    });

    eventSource.onerror = () => {
      console.error('SSE connection error, will auto-reconnect...');
    };

    return () => eventSource.close();
  }, [symbols.join(',')]);

  return prices;
}

// components/StockDashboard.tsx
function StockDashboard() {
  const prices = useStockStream(['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA']);

  return (
    <div className="grid grid-cols-5 gap-4">
      {Array.from(prices.entries()).map(([symbol, price]) => (
        <div key={symbol} className="p-4 rounded-lg bg-gray-800">
          <h3 className="text-lg font-bold">{symbol}</h3>
          <p className="text-2xl">${price.price.toFixed(2)}</p>
          <p className={price.change >= 0 ? 'text-green-400' : 'text-red-400'}>
            {price.change >= 0 ? '+' : ''}{price.change.toFixed(2)}%
          </p>
        </div>
      ))}
    </div>
  );
}

확장 전 (1 인스턴스):
  WebFlux Instance-1 ← Kafka Consumer (P0, P1, P2, P3, P4, P5)

확장 후 (3 인스턴스):
  WebFlux Instance-1 ← Kafka Consumer (P0, P1)  ← SSE Client 1~1000
  WebFlux Instance-2 ← Kafka Consumer (P2, P3)  ← SSE Client 1001~2000
  WebFlux Instance-3 ← Kafka Consumer (P4, P5)  ← SSE Client 2001~3000

핵심: Kafka 파티션 수 = 최대 병렬 Consumer 수
     파티션 키(symbol)로 같은 종목은 같은 파티션에 순서 보장

# docker-compose-monitoring.yml
services:
  prometheus:
    image: prom/prometheus:v2.51.0
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:10.4.0
    ports:
      - '3000:3000'
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

  kafka-exporter:
    image: danielqsj/kafka-exporter:latest
    command: ['--kafka.server=kafka:9092']
    ports:
      - '9308:9308'

// Micrometer 메트릭 추가
@Component
@RequiredArgsConstructor
public class StockMetrics {

    private final MeterRegistry registry;
    private final AtomicLong connectedClients = new AtomicLong(0);

    @PostConstruct
    void init() {
        Gauge.builder("stock.stream.connected_clients", connectedClients, AtomicLong::get)
            .description("Number of connected SSE clients")
            .register(registry);
    }

    public void recordPricePublished(String symbol) {
        registry.counter("stock.price.published",
            "symbol", symbol).increment();
    }

    public void recordLatency(String symbol, long latencyMs) {
        registry.timer("stock.price.latency",
            "symbol", symbol)
            .record(Duration.ofMillis(latencyMs));
    }

    public void clientConnected() { connectedClients.incrementAndGet(); }
    public void clientDisconnected() { connectedClients.decrementAndGet(); }
}