cursus

Cluster Consumer Architecture

Consumer가 클러스터 환경에서 메시지를 소비하기 위한 FindCoordinator 기반 라우팅 아키텍처.

Overview

Consumer group commands are routed to the group coordinator, while CONSUME and STREAM are routed to the partition leader. The broker-owned committed nextOffset remains authoritative across both routes.

Command Routing

flowchart TB
    Consumer[Consumer SDK]

    subgraph "1. Coordinator Discovery"
        ANY[Any Broker]
        FC[FIND_COORDINATOR]
        Consumer -->|"①"| ANY
        ANY -->|"group=my-group"| FC
        FC -->|"OK host=H port=P"| Consumer
    end

    subgraph "2. Group Commands → Coordinator"
        COORD[Coordinator Broker]
        Consumer -->|"②"| COORD
        COORD --- JOIN[JOIN_GROUP]
        COORD --- SYNC[SYNC_GROUP]
        COORD --- HB[HEARTBEAT]
        COORD --- LEAVE[LEAVE_GROUP]
        COORD --- FETCH[FETCH_OFFSET]
        COORD --- COMMIT[COMMIT_OFFSET]
        COORD --- BATCH[BATCH_COMMIT]
    end

    subgraph "3. Data Commands → Partition Leader"
        META[METADATA]
        Consumer -->|"③"| ANY
        ANY -->|"leaders=H1:P1,H2:P2"| META
    end

    subgraph "4. Consume from Leaders"
        L1[Partition 0 Leader]
        L2[Partition 1 Leader]
        Consumer -->|"CONSUME P0"| L1
        Consumer -->|"CONSUME P1"| L2
    end

Command Routing Table

Command	Target	Error Handling
FIND_COORDINATOR	Any broker	Retry with next broker
JOIN_GROUP	Coordinator	NOT_COORDINATOR → re-discover
SYNC_GROUP	Coordinator	NOT_COORDINATOR → re-discover
LEAVE_GROUP	Coordinator	NOT_COORDINATOR → re-discover
HEARTBEAT	Coordinator	NOT_COORDINATOR → re-discover
FETCH_OFFSET	Coordinator	NOT_COORDINATOR → re-discover
COMMIT_OFFSET	Coordinator	NOT_COORDINATOR → re-discover
BATCH_COMMIT	Coordinator	NOT_COORDINATOR → re-discover
CONSUME	Partition Leader	NOT_LEADER → update leader cache
STREAM	Partition Leader	NOT_LEADER → update leader cache

Consumer Lifecycle (Cluster)

The Go SDK exposes one lifecycle through Consumer.State():

new -> running <-> rebalancing -> closing -> closed

Close wins over every in-progress transition and is idempotent. The root context owns the rebalance monitor, commit worker, metadata/heartbeat loops, partition readers, and handler workers. Rebalance cancels and joins the old assignment before installing its replacement.

Every assignment has a local monotonically increasing generation in addition to the broker group generation. Poll, stream, handler, and commit paths carry that local generation. Once rebalancing begins, work from the prior assignment is rejected with sdk.ErrConsumerRebalancing before another commit is sent. This prevents a late worker from mutating the newly installed assignment.

sequenceDiagram
    participant C as Consumer
    participant B as Any Broker
    participant CO as Coordinator
    participant PL as Partition Leader

    C->>B: FIND_COORDINATOR group=G
    B-->>C: OK host=H port=P

    C->>CO: JOIN_GROUP topic=T group=G member=M
    CO-->>C: OK generation=1 member=M-1234 assignments=[0,1]

    C->>CO: SYNC_GROUP topic=T group=G member=M-1234 generation=1
    CO-->>C: OK generation=1 member=M-1234 assignments=[0,1]

    C->>CO: FETCH_OFFSET topic=T partition=0 group=G
    CO-->>C: OK offset=0

    C->>B: METADATA topic=T
    B-->>C: OK leaders=H1:P1,H2:P2

    loop Poll Loop
        C->>PL: CONSUME topic=T partition=0 offset=<committed> member=M-1234 group=G generation=1 isolation=read_committed
        PL-->>C: batch(messages)
    end

    par Heartbeat (every 3s)
        C->>CO: HEARTBEAT topic=T group=G member=M-1234 generation=1
        CO-->>C: OK
    end

Transient connection failures do not require a new member. The SDK first retries JOIN_GROUP with the broker-assigned member ID and generation. A successful resume preserves assignments and does not advance the generation. A stale generation is synchronized for an existing member; member_not_found causes a fresh join. When group coordinator ownership changes, the new owner waits one session timeout before expiring members, while clients rediscover it and resume heartbeats.

Resume And Isolation Rules

• After every assignment or rejoin, fetch FETCH_OFFSET for each assigned partition before reading.
• When a committed offset exists, the data path starts there even if the request carries a lower explicit offset.
• If no offset exists, autoOffsetReset=earliest starts at the earliest retained offset, latest starts at the next committed tail, and error surfaces the gap.
• Commit only lastProcessedOffset + 1. Regressions fail and do not change broker state.
• read_committed is the SDK default and hides unresolved/aborted transaction records. read_uncommitted returns the raw committed log, including transaction control records.
• A graceful STREAM_CONTROL type=CLOSE is a terminator. A socket loss without that frame is retryable and must resume through the group offset contract.

Error Recovery

flowchart TD
    SEND[Send Command]
    SEND --> CHECK{Response?}

    CHECK -->|NOT_COORDINATOR| REDISC[Re-discover Coordinator]
    REDISC --> PARSE{host/port in response?}
    PARSE -->|Yes| UPDATE[Update coordinator addr]
    PARSE -->|No| FIND[FIND_COORDINATOR]
    UPDATE --> RETRY[Retry command]
    FIND --> RETRY

    CHECK -->|NOT_LEADER leader=addr| ULEAD[Update partition leader cache]
    ULEAD --> RETRY

    CHECK -->|OK / data| DONE[Process response]

Implementation Details

Go SDK

• findCoordinator() — sends FIND_COORDINATOR via ConnectWithFailover
• getCoordinatorConn() — connects to coordinator, falls back to findCoordinator on failure
• fetchMetadata() — sends METADATA topic=<topic>, populates partitionLeaders map
• getPartitionLeaderAddr(partitionID) / updatePartitionLeader(partitionID, addr) — thread-safe leader cache
• ensureConnection() — prefers partition leader address, falls back to any broker
• handleBrokerError() — parses NOT_LEADER, updates partition leader, triggers rebalance on GEN_MISMATCH
• handleNotCoordinator() — re-discovers coordinator from response or via FIND_COORDINATOR
• joinGroup() — resumes the current member and generation before creating a fresh member
• fetchOffset() — uses a bounded, per-request coordinator connection and parses OK offset=<N>
• ReadIsolation — sends explicit isolation=read_committed|read_uncommitted on polling and streaming reads
• offset commits — send the broker-assigned member and generation, reject regressions, and stop/rejoin on fencing errors
• Consumer.State() — reports new, running, rebalancing, closing, or closed
• local assignment generations — fence stale fetch and commit workers across rebalances

This site is open source. Improve this page.