Producer and Consumer Clients — Leader Discovery, Reconnection, and Offset Tracking

In this section you'll build the client-side libraries that produce and consume records, and the small piece of server state that remembers where each consumer left off. Three packages are involved, and all three happen to be named client, so it's worth being precise about which is which up front: the top-level client/ package holds discovery and retry logic shared by everyone; producer/client/ and consumer/client/ each hold a package also named client that builds a topic-aware, self-healing wrapper on top of that shared logic. producer/cmd/ and consumer/cmd/ are the CLI programs that tie a client.Client to stdin/stdout, and broker/consumer/ is the server-side package that persists committed offsets.

Step 1: Build a RemoteClient for admin and discovery calls

RemoteClient wraps a single connection and exposes the cluster-level RPCs: create/delete a topic, find a leader, list topics.

// client/rpc.go
package client

type RemoteClient struct {
    tc *transport.TransportClient
}

func NewRemoteClient(addr string) (*RemoteClient, error) {
    tc, err := transport.DialWithFallback(addr)
    if err != nil {
        return nil, err
    }
    return &RemoteClient{tc: tc}, nil
}

func (c *RemoteClient) Close() error {
    return c.tc.Close()
}

func (c *RemoteClient) CreateTopic(ctx context.Context, req *protocol.CreateTopicRequest) (*protocol.CreateTopicResponse, error) {
    resp, err := c.tc.Call(*req)
    if err != nil {
        return nil, err
    }
    r := resp.(protocol.CreateTopicResponse)
    return &r, nil
}

// FindLeader asks a node which RPC address is currently the leader for the given topic.
func (c *RemoteClient) FindTopicLeader(ctx context.Context, req *protocol.FindTopicLeaderRequest) (*protocol.FindTopicLeaderResponse, error) {
    resp, err := c.tc.Call(*req)
    if err != nil {
        return nil, err
    }
    r := resp.(protocol.FindTopicLeaderResponse)
    return &r, nil
}

// GetRaftLeader asks a node for the Raft (metadata) leader RPC address. Use this before create-topic.
func (c *RemoteClient) FindRaftLeader(ctx context.Context, req *protocol.FindRaftLeaderRequest) (*protocol.FindRaftLeaderResponse, error) {
    resp, err := c.tc.Call(*req)
    if err != nil {
        return nil, err
    }
    r := resp.(protocol.FindRaftLeaderResponse)
    return &r, nil
}

DeleteTopic and ListTopics follow the identical shape: dereference the request, call tc.Call, type-assert the response. FindTopicLeader, FindRaftLeader, and ListTopics can be answered by any node, since they just read locally-held cluster metadata. CreateTopic and DeleteTopic must land on the Raft leader specifically, since they propose a metadata change — which is why callers reach for FindRaftLeader first when they need to mutate topics.

Step 2: Try each bootstrap address in turn with TryAddrs

A client doesn't know in advance which broker is reachable or which one leads a given topic, so it needs to try a list of seed addresses until one answers:

// client/bootstrap.go
package client

var errNoConnection = errors.New("could not connect to any address")

// TryAddrs tries each address in addrs in order. For each address it creates a RemoteClient,
// calls fn(client), then closes the client. If fn returns a non-empty result and nil error,
// TryAddrs returns that result. An empty result with a nil error is treated the same as an
// error (fn's contract is "non-empty result on success") — it moves on to the next address
// rather than returning ("", nil), which would look like success to a caller that only
// checks err. If fn returns a real error and ShouldReconnect(err) is true, it also tries the
// next address; otherwise it returns the error.
func TryAddrs(ctx context.Context, addrs []string, fn func(*RemoteClient) (string, error)) (string, error) {
    var lastErr error
    for _, addr := range addrs {
        addr = strings.TrimSpace(addr)
        if addr == "" {
            continue
        }
        c, err := NewRemoteClient(addr)
        if err != nil {
            lastErr = err
            continue
        }
        result, err := fn(c)
        _ = c.Close()
        if err == nil && result != "" {
            return result, nil
        }
        if err == nil {
            err = fmt.Errorf("empty result returned from %s", addr)
            lastErr = err
            continue
        }
        lastErr = err
        if ShouldReconnect(err) {
            continue
        }
        return "", err
    }
    if lastErr != nil {
        return "", lastErr
    }
    return "", errNoConnection
}

Each address gets a fresh, short-lived RemoteClient that's closed right after the callback runs — TryAddrs is for one-shot discovery calls, not for holding a connection open. The three ways a given address can end the loop are worth spelling out: the callback returns a non-empty result (done), it returns an error ShouldReconnect recognizes as connection- or leader-shaped (move on to the next address), or it returns anything else (stop and surface that error — retrying a different broker won't fix, say, a malformed request).

Step 3: Resolve and re-resolve the topic leader

ResolveTopicLeader turns TryAddrs into a resolver that specifically asks for a topic's leader, and ReconnectBackoff turns any connect function into a retry loop with linear backoff. Both are used by the producer and consumer clients to (re)connect after a failover:

// client/leader.go
package client

// ResolveTopicLeader returns a resolver that discovers topic's current leader RPC
// address by trying each of bootstrapAddrs in turn (see TryAddrs) — the leader lookup
// producer/client.Client and consumer/client.Client use to (re)connect.
func ResolveTopicLeader(bootstrapAddrs []string, topic string) func(ctx context.Context) (string, error) {
    return func(ctx context.Context) (string, error) {
        return TryAddrs(ctx, bootstrapAddrs, func(c *RemoteClient) (string, error) {
            resp, err := c.FindTopicLeader(ctx, &protocol.FindTopicLeaderRequest{Topic: topic})
            if err != nil {
                return "", err
            }
            if resp.LeaderAddr == "" {
                return "", fmt.Errorf("empty leader address for topic %s", topic)
            }
            return resp.LeaderAddr, nil
        })
    }
}

// ReconnectBackoff calls connect repeatedly, backing off linearly (attempt * 500ms)
// between tries, until it succeeds, ctx is done, or maxAttempts is reached. This is the
// shared "leader moved, redial" retry policy for producer/client.Client and
// consumer/client.Client — the same shape both CLIs used to hand-roll independently.
func ReconnectBackoff(ctx context.Context, maxAttempts int, connect func() error) error {
    var lastErr error
    for attempt := 0; attempt < maxAttempts; attempt++ {
        if err := ctx.Err(); err != nil {
            return err
        }
        if err := connect(); err == nil {
            return nil
        } else {
            lastErr = err
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(time.Duration(attempt+1) * 500 * time.Millisecond):
        }
    }
    return fmt.Errorf("reconnect failed after %d attempts: %w", maxAttempts, lastErr)
}

ResolveTopicLeader(addrs, topic) returns a plain func(ctx) (string, error) rather than doing the lookup directly, so ReconnectBackoff can wrap it in a retry loop without either function needing to know about the other's retry policy. The backoff itself is linear and capped at 10 attempts by callers (500ms, 1s, 1.5s, ...) — enough to ride out a short election without hanging indefinitely.

Step 4: Two questions, two functions — ShouldReconnect and RetryTopicNotReady

Every RPC failure a producer or consumer client sees needs an answer to one of two different questions: "is this connection/leader actually gone, or is the topic just not open on this node yet?" client/reconnect.go re-exports the protocol-level answer to the first question:

// client/reconnect.go
package client

// ShouldReconnect reports whether the caller should re-resolve the leader and create a new client.
// Use after Produce, Fetch, or other RPC calls fail; if true, reconnect to the current leader and retry.
func ShouldReconnect(err error) bool {
    return protocol.ShouldReconnect(err)
}

client/retry.go answers the second question with its own retry loop, on the same connection:

// client/retry.go
package client

const (
    topicNotReadyBackoff = 50 * time.Millisecond
    topicNotReadyMaxWait = 2 * time.Second
)

// RetryTopicNotReady retries fn while it fails with protocol.IsTopicNotReady — the
// brief window after CreateTopic (or a leader change) before the target node's own
// periodic metadata reconciliation has opened the local log — backing off between
// attempts, until fn succeeds, ctx is done, or the retry budget is spent. Any other
// error (including a genuinely nonexistent topic, which looks identical from the
// client's perspective) is returned once that budget runs out, same tradeoff Kafka's
// own producer/consumer clients make with max.block.ms.
func RetryTopicNotReady(ctx context.Context, fn func() error) error {
    deadline := time.Now().Add(topicNotReadyMaxWait)
    for {
        err := fn()
        if err == nil || !protocol.IsTopicNotReady(err) || !time.Now().Before(deadline) {
            return err
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(topicNotReadyBackoff):
        }
    }
}

Together these four files (rpc.go, bootstrap.go, leader.go, reconnect.go, retry.go) give every client three distinct, ordered layers of retry:

1. Same connection, same node — RetryTopicNotReady. Right after CreateTopic, the node a client happens to be talking to may not have opened the topic's local log yet. That resolves itself in milliseconds, so the client just waits (50ms steps, up to 2s) and retries the identical request on the identical connection.
2. New connection, same cluster — ShouldReconnect + reconnect. If the error means the leader actually changed or the connection genuinely died, retrying the same node is pointless. The client closes its connection, calls ResolveTopicLeader to find the new leader, and uses ReconnectBackoff to retry that discovery up to 10 times.
3. New cluster member — TryAddrs inside ResolveTopicLeader. Leader discovery itself doesn't assume the first bootstrap address is still up; it walks the whole address list, so a client started with several seed addresses survives losing any one of them.

This is written once in client/ instead of once per CLI because three different callers need exactly this behavior: the producer client, the consumer client, and (as covered in the replication chapter) the broker's own replication engine, which reuses consumer/client.ConsumerClient internally to pull data from other leaders. Writing "is this error worth reconnecting over" three separate times would have meant three separate chances to get the edge cases wrong.

Step 5: Dial once, with a Docker-hostname fallback

Every client type above (RemoteClient, and the producer/consumer clients below) opens its connection through one shared function instead of calling net.Dial directly:

// api/transport/transport.go

// DialWithFallback is like Dial, but when addr is an unresolvable Docker-internal
// hostname (e.g. "node1:9092" returned by FindLeader/FindRaftLeader from inside the
// cluster's network), it falls back to 127.0.0.1:<port>, which works when the caller
// is running outside Docker with the port mapped to the host.
//
// Every RPC client (admin, producer, consumer, and internal replication, which uses
// the consumer client) dials through this one function so the fallback behavior lives
// in a single place instead of being copy-pasted per client.
func DialWithFallback(addr string) (*TransportClient, error) {
    tc, err := Dial(addr)
    if err != nil {
        if strings.Contains(err.Error(), "no such host") {
            if host, port, splitErr := net.SplitHostPort(addr); splitErr == nil && strings.HasPrefix(host, "node") && port != "" {
                fallback := net.JoinHostPort("127.0.0.1", port)
                if tc2, err2 := Dial(fallback); err2 == nil {
                    return tc2, nil
                }
            }
        }
        return nil, err
    }
    return tc, nil
}

The scenario this handles: FindTopicLeader returns an address like node2:9094 — resolvable from inside the Docker network the cluster runs in, but not from a CLI running on the host. If the hostname looks like a cluster node name (strings.HasPrefix(host, "node")) and plain Dial failed with a DNS lookup error, DialWithFallback retries against 127.0.0.1 on the same port, which works as long as Docker Compose maps that port to the host. Because every client dials through this one function, none of them need to special-case Docker locally.

Step 6: ProducerClient — one connection to the topic leader

producer/client/producer.go defines the low-level producer: a single connection, Produce and ProduceBatch.

// producer/client/producer.go
package client

type ProducerClient struct {
    tc *transport.TransportClient
}

func NewProducerClient(addr string) (*ProducerClient, error) {
    tc, err := transport.DialWithFallback(addr)
    if err != nil {
        return nil, err
    }
    return &ProducerClient{tc: tc}, nil
}

func (c *ProducerClient) Close() error {
    return c.tc.Close()
}

func (c *ProducerClient) Produce(ctx context.Context, req *protocol.ProduceRequest) (*protocol.ProduceResponse, error) {
    var resp protocol.ProduceResponse
    err := toplevelclient.RetryTopicNotReady(ctx, func() error {
        r, err := c.tc.Call(*req)
        if err != nil {
            return err
        }
        resp = r.(protocol.ProduceResponse)
        return nil
    })
    if err != nil {
        return nil, err
    }
    return &resp, nil
}

func (c *ProducerClient) ProduceBatch(ctx context.Context, req *protocol.ProduceBatchRequest) (*protocol.ProduceBatchResponse, error) {
    var resp protocol.ProduceBatchResponse
    err := toplevelclient.RetryTopicNotReady(ctx, func() error {
        r, err := c.tc.Call(*req)
        if err != nil {
            return err
        }
        resp = r.(protocol.ProduceBatchResponse)
        return nil
    })
    if err != nil {
        return nil, err
    }
    return &resp, nil
}

ProducerClient is deliberately thin: it knows how to talk to exactly one address and applies exactly the first retry tier (RetryTopicNotReady) from Step 4. It has no idea what a leader change is — that's the next layer up.

Step 7: producer/client.Client — leader discovery and automatic reconnect

producer/client/client.go wraps ProducerClient in a topic-aware Client that finds the leader, holds the current connection behind a mutex, and reconnects on failover without the caller noticing:

// producer/client/client.go
package client

// Client is a topic-aware producer: it discovers the current topic leader and
// reconnects automatically on failover or leader change, so callers never see address
// resolution or reconnect logic — mirrors how KafkaProducer.send() hides
// NetworkClient's connection/leader-tracking from the caller. ProducerClient (which
// this wraps) already retries same-connection for the brief post-create/leader-change
// window (see ProducerClient.Produce); Client adds the second tier Kafka's client also
// has: reconnecting to a different broker when the leader actually moved or the
// connection died.
type Client struct {
    mu             sync.Mutex
    bootstrapAddrs []string
    topic          string
    pc             *ProducerClient
    addr           string

    // OnReconnect, if set, is called with the new leader address after Client
    // transparently reconnects following a failover/leader change. Optional — purely
    // for callers (e.g. a CLI) that want to surface status; Client never requires it.
    OnReconnect func(addr string)
}

// NewClient resolves topic's current leader among bootstrapAddrs and connects to it.
func NewClient(ctx context.Context, bootstrapAddrs []string, topic string) (*Client, error) {
    c := &Client{bootstrapAddrs: bootstrapAddrs, topic: topic}
    if err := c.connect(ctx); err != nil {
        return nil, err
    }
    return c, nil
}

func (c *Client) connect(ctx context.Context) error {
    addr, err := toplevelclient.ResolveTopicLeader(c.bootstrapAddrs, c.topic)(ctx)
    if err != nil {
        return err
    }
    pc, err := NewProducerClient(addr)
    if err != nil {
        return err
    }
    c.mu.Lock()
    old := c.pc
    c.pc, c.addr = pc, addr
    c.mu.Unlock()
    if old != nil {
        _ = old.Close()
    }
    return nil
}

func (c *Client) reconnect(ctx context.Context) error {
    if err := toplevelclient.ReconnectBackoff(ctx, 10, func() error { return c.connect(ctx) }); err != nil {
        return err
    }
    if c.OnReconnect != nil {
        c.OnReconnect(c.LeaderAddr())
    }
    return nil
}

// Send produces one record, transparently reconnecting to the new leader and retrying
// on failover or connection failure, until it succeeds or ctx is done.
func (c *Client) Send(ctx context.Context, value []byte, acks protocol.AckMode) (uint64, error) {
    for {
        c.mu.Lock()
        pc := c.pc
        c.mu.Unlock()
        resp, err := pc.Produce(ctx, &protocol.ProduceRequest{Topic: c.topic, Value: value, Acks: acks})
        if err == nil {
            return resp.Offset, nil
        }
        if !toplevelclient.ShouldReconnect(err) {
            return 0, err
        }
        if rerr := c.reconnect(ctx); rerr != nil {
            return 0, fmt.Errorf("produce failed (%v), reconnect failed: %w", err, rerr)
        }
        if ctxErr := ctx.Err(); ctxErr != nil {
            return 0, ctxErr
        }
    }
}

SendBatch has the identical shape, calling pc.ProduceBatch instead of pc.Produce. Notice the connection swap in connect: a new ProducerClient is dialed before the mutex is taken, the pointer swap under the lock is instantaneous, and the old connection is closed only after the swap — so a concurrent Send reading c.pc never observes a half-torn-down client. acks is a protocol.AckMode (AckNone, AckLeader, or AckAll — defined in api/protocol/types.go and covered on the wire-protocol page); Client.Send/SendBatch just forward whatever mode the caller passes through to ProduceRequest.Acks.

Step 8: Wire the producer CLI to producer/client.Client

producer/cmd/main.go is a cobra program with one connect subcommand: resolve the leader once via NewClient, then loop reading stdin and calling Send.

// producer/cmd/main.go
rootCmd.PersistentFlags().StringVar(&addrs, "addrs", "127.0.0.1:9094", "Comma-separated RPC addresses to try for discovery (tried in order until one connects)")
rootCmd.PersistentFlags().StringVar(&topic, "topic", "", "topic name (required)")
rootCmd.PersistentFlags().Int32Var(&acks, "acks", int32(protocol.AckLeader), "acks: 0=none,1=leader,2=all")
rootCmd.MarkPersistentFlagRequired("topic")

connectCmd := &cobra.Command{
    Use:   "connect",
    Short: "Connect to the topic leader and produce messages from stdin",
    RunE: func(cmd *cobra.Command, args []string) error {
        ackMode := protocol.AckMode(acks)
        if ackMode != protocol.AckNone && ackMode != protocol.AckLeader && ackMode != protocol.AckAll {
            ackMode = protocol.AckLeader
        }

        ctx := context.Background()

        // NewClient discovers the topic leader among --addrs and connects; Client
        // itself handles re-discovery and reconnecting on failover from here on,
        // so this command never has to.
        connectCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
        c, err := producerclient.NewClient(connectCtx, addrList(), topic)
        cancel()
        if err != nil {
            return err
        }
        defer c.Close()
        c.OnReconnect = func(addr string) {
            fmt.Fprintf(os.Stderr, "reconnected to topic %q leader at %s\n", topic, addr)
        }

        scanner := bufio.NewScanner(os.Stdin)
        for scanner.Scan() {
            line := strings.TrimRight(scanner.Text(), "\r\n")
            if line == "" {
                continue
            }
            msgCtx, cancelMsg := context.WithTimeout(ctx, 10*time.Second)
            offset, err := c.Send(msgCtx, []byte(line), ackMode)
            cancelMsg()
            if err != nil {
                return err
            }
            fmt.Printf("offset=%d\n", offset)
        }
        return scanner.Err()
    },
}

The CLI itself contains no discovery or reconnect logic at all — every line typed at stdin is just handed to c.Send, and Client handles whatever happens underneath. --addrs defaults to 127.0.0.1:9094, --topic is required, and --acks takes the numeric AckMode (0/1/2, default 1 — AckLeader).

Step 9: ConsumerClient — one connection for fetch and offset commit

consumer/client/consumer.go is the low-level counterpart to ProducerClient: one connection, plus Fetch, FetchBatch, CommitOffset, and FetchOffset.

// consumer/client/consumer.go
package client

type ConsumerClient struct {
    tc            *transport.TransportClient
    ReplicaNodeID string // when set, Fetch uses read-uncommitted (replication); otherwise read up to HW
    noRetry       bool
}

func NewConsumerClient(addr string) (*ConsumerClient, error) {
    tc, err := transport.DialWithFallback(addr)
    if err != nil {
        return nil, err
    }
    return &ConsumerClient{tc: tc}, nil
}

// SetReplicaNodeID sets the client to replication mode: Fetch requests will include
// ReplicaNodeID so the leader uses ReadUncommitted and records replica LEO.
func (c *ConsumerClient) SetReplicaNodeID(id string) {
    c.ReplicaNodeID = id
}

// DisableTopicNotReadyRetry turns off Fetch/FetchBatch's automatic retry on
// CodeTopicNotFound/CodeNotTopicLeader. Internal replication (topic.ReplicateFromLeader)
// already retries at a coarser granularity — abort this leader, let the next replication
// tick try again — so the two would otherwise stack: a stalled leader could add the full
// client-side retry budget as extra blocking latency inside a single replication tick on
// top of the tick-level retry that already handles it.
func (c *ConsumerClient) DisableTopicNotReadyRetry() {
    c.noRetry = true
}

func (c *ConsumerClient) Fetch(ctx context.Context, req *protocol.FetchRequest) (*protocol.FetchResponse, error) {
    reqCopy := *req
    if c.ReplicaNodeID != "" {
        reqCopy.ReplicaNodeID = c.ReplicaNodeID
    }
    var resp protocol.FetchResponse
    call := func() error {
        r, err := c.tc.Call(reqCopy)
        if err != nil {
            return err
        }
        resp = r.(protocol.FetchResponse)
        return nil
    }
    var err error
    if c.noRetry {
        err = call()
    } else {
        err = toplevelclient.RetryTopicNotReady(ctx, call)
    }
    if err != nil {
        return nil, err
    }
    return &resp, nil
}

FetchBatch mirrors Fetch exactly; CommitOffset and FetchOffset are plain single-call wrappers with no retry logic of their own (an offset commit that fails just gets reported to the caller — see Step 11). Both Fetch and FetchBatch copy the request before stamping ReplicaNodeID onto it, so they never mutate a struct the caller still holds a reference to.

ReplicaNodeID and DisableTopicNotReadyRetry exist for exactly one caller outside the consumer CLI: the broker's own replication engine (broker/topic/replication.go, covered in the replication chapter) imports this same ConsumerClient type to pull data from other nodes' leaders. Setting ReplicaNodeID marks a fetch as a replica read (server-side: ReadUncommitted, up to LEO instead of the high watermark) and records the caller's LEO for ISR tracking; DisableTopicNotReadyRetry turns off this file's own retry loop because the replication tick loop already retries at a coarser grain (abandon this leader for the tick, try again next tick), and stacking a second retry budget underneath would only add latency.

Step 10: consumer/client.Client — Poll, Commit, and FetchCommittedOffset

consumer/client/client.go layers the same discovery-and-reconnect behavior from Step 7 on top of ConsumerClient, exposing Poll (fetch-and-wait), Commit, and FetchCommittedOffset:

// consumer/client/client.go
package client

type Client struct {
    mu             sync.Mutex
    bootstrapAddrs []string
    topic, id      string
    cc             *ConsumerClient
    addr           string
    OnReconnect    func(addr string)
}

// NewClient resolves topic's current leader among bootstrapAddrs and connects to it.
// id identifies this consumer for offset commit/fetch.
func NewClient(ctx context.Context, bootstrapAddrs []string, topic, id string) (*Client, error) {
    c := &Client{bootstrapAddrs: bootstrapAddrs, topic: topic, id: id}
    if err := c.connect(ctx); err != nil {
        return nil, err
    }
    return c, nil
}

// Poll fetches the next record at offset. While the leader reports "caught up, nothing
// new yet" (CodeReadOffset), it sleeps pollInterval and retries. On failover or
// connection failure it reconnects and retries. Returns when a record is available,
// ctx is done, or a non-retriable error occurs.
func (c *Client) Poll(ctx context.Context, offset uint64, pollInterval time.Duration) (*protocol.LogEntry, error) {
    for {
        c.mu.Lock()
        cc := c.cc
        c.mu.Unlock()
        resp, err := cc.Fetch(ctx, &protocol.FetchRequest{Id: c.id, Topic: c.topic, Offset: offset})
        if err == nil {
            if resp.Entry != nil {
                return resp.Entry, nil
            }
            if werr := waitOrDone(ctx, pollInterval); werr != nil {
                return nil, werr
            }
            continue
        }

        var rpcErr *protocol.RPCError
        if errors.As(err, &rpcErr) && rpcErr.Code == protocol.CodeReadOffset {
            if werr := waitOrDone(ctx, pollInterval); werr != nil {
                return nil, werr
            }
            continue
        }

        if !toplevelclient.ShouldReconnect(err) {
            return nil, err
        }
        if rerr := c.reconnect(ctx); rerr != nil {
            return nil, fmt.Errorf("fetch failed (%v), reconnect failed: %w", err, rerr)
        }
        if ctxErr := ctx.Err(); ctxErr != nil {
            return nil, ctxErr
        }
    }
}

// Commit commits offset for this consumer's id/topic, reconnecting and retrying once
// on failover or connection failure.
func (c *Client) Commit(ctx context.Context, offset uint64) error {
    for {
        c.mu.Lock()
        cc := c.cc
        c.mu.Unlock()
        _, err := cc.CommitOffset(ctx, &protocol.CommitOffsetRequest{Id: c.id, Topic: c.topic, Offset: offset})
        if err == nil {
            return nil
        }
        if !toplevelclient.ShouldReconnect(err) {
            return err
        }
        if rerr := c.reconnect(ctx); rerr != nil {
            return fmt.Errorf("commit offset failed (%v), reconnect failed: %w", err, rerr)
        }
        if ctxErr := ctx.Err(); ctxErr != nil {
            return ctxErr
        }
    }
}

// FetchCommittedOffset returns the last committed offset for this consumer's id/topic.
func (c *Client) FetchCommittedOffset(ctx context.Context) (uint64, error) {
    // Same reconnect-and-retry shape as Commit, calling cc.FetchOffset instead.
    ...
}

Poll folds three outcomes into one call: a record is ready (return it), the leader reports CodeReadOffset meaning the consumer has caught up to the high watermark (sleep pollInterval, poll again), or the connection/leader is actually gone (ShouldReconnect, reconnect, retry). A caller never needs to distinguish "no data yet" from "reconnecting" — both look like Poll blocking a little longer.

Step 11: Wire the consumer CLI to consumer/client.Client

consumer/cmd/main.go resolves a starting offset, then loops Poll → print → Commit:

// consumer/cmd/main.go
rootCmd.PersistentFlags().StringVar(&addrs, "addrs", "127.0.0.1:9092", "Comma-separated RPC addresses to try for discovery (tried in order until one connects)")
rootCmd.PersistentFlags().StringVar(&id, "id", "default", "consumer id")
rootCmd.PersistentFlags().StringVar(&topic, "topic", "", "topic name (required)")
rootCmd.PersistentFlags().Uint64Var(&offset, "offset", 0, "start from specific offset (default: resume from last committed)")
rootCmd.PersistentFlags().BoolVar(&fromBeginning, "from-beginning", false, "start from offset 0 instead of last committed offset")

// ...inside the connect command's RunE...
offsetExplicitlySet := cmd.Flags().Changed("offset")
startOffset := offset

if fromBeginning {
    startOffset = 0
} else if offsetExplicitlySet {
    fmt.Fprintf(os.Stderr, "Starting from offset %d (explicitly specified)\n", startOffset)
} else {
    fetchCtx, fetchCancel := context.WithTimeout(ctx, 5*time.Second)
    committed, err := c.FetchCommittedOffset(fetchCtx)
    fetchCancel()
    if err == nil && committed > 0 {
        startOffset = committed
        fmt.Fprintf(os.Stderr, "Resuming from offset %d (last committed)\n", startOffset)
    } else {
        startOffset = 0
    }
}

currentOffset := startOffset
pollInterval := 500 * time.Millisecond

for {
    entry, err := c.Poll(ctx, currentOffset, pollInterval)
    if err != nil {
        return err
    }
    fmt.Printf("%d\t%s\n", entry.Offset, string(entry.Value))
    currentOffset = entry.Offset + 1

    commitCtx, commitCancel := context.WithTimeout(ctx, 5*time.Second)
    if err := c.Commit(commitCtx, currentOffset); err != nil {
        fmt.Fprintf(os.Stderr, "warning: commit offset %d failed: %v\n", currentOffset, err)
    }
    commitCancel()
}

The starting offset is resolved in a fixed priority order: --from-beginning wins outright (start at 0); otherwise an explicitly-set --offset flag (checked via cmd.Flags().Changed("offset"), so an unset --offset doesn't shadow the committed lookup with its zero value) wins; otherwise the client asks the server for the last committed offset via FetchCommittedOffset and falls back to 0 if none is found. Every consumed record is committed immediately after being printed — a commit failure is only logged, not fatal, since the next successful commit will simply move the stored offset forward again.

Step 12: Track committed offsets on the server with ConsumerManager

Every commit needs somewhere durable to live on the broker side, so a restart doesn't lose consumer positions. broker/consumer/consumer.go defines ConsumerManager, which keeps an in-memory cache backed by an append-only log:

// broker/consumer/consumer.go
package consumer

type ConsumerManager struct {
    mu          sync.RWMutex
    offsetCache map[string]map[string]uint64
    offsetLog   *log.Log
    recoverOnce sync.Once
    recoverErr  error
}

func NewConsumerManager(baseDir string) (*ConsumerManager, error) {
    offsetLog, err := log.NewLog(filepath.Join(baseDir, "__consumer_offsets__.log"))
    if err != nil {
        return nil, err
    }
    cm := &ConsumerManager{
        offsetCache: make(map[string]map[string]uint64),
        offsetLog:   offsetLog,
    }
    // Recover offsets eagerly at startup instead of per-request.
    if err := cm.Recover(); err != nil {
        return nil, fmt.Errorf("consumer offset recovery: %w", err)
    }
    return cm, nil
}

func (c *ConsumerManager) CommitOffset(id string, topic string, offset uint64) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    if _, ok := c.offsetCache[id]; !ok {
        c.offsetCache[id] = make(map[string]uint64)
    }
    c.offsetCache[id][topic] = offset
    _, err := c.offsetLog.Append([]byte(fmt.Sprintf("%s,%s,%d", id, topic, offset)))
    return err
}

func (c *ConsumerManager) GetOffset(id string, topic string) (uint64, error) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    topicMap, ok := c.offsetCache[id]
    if !ok {
        return 0, ErrOffsetNotFoundForID(id, topic)
    }
    off, ok := topicMap[topic]
    if !ok {
        return 0, ErrOffsetNotFoundForID(id, topic)
    }
    return off, nil
}

offsetCache is a nested map, consumerID → topic → offset, guarded by an RWMutex so reads don't block each other. Every commit appends a CSV line ("consumerID,topic,offset") to offsetLog — an instance of the same append-only broker/log.Log type used for topic data — before updating the cache, so the write survives a crash between the append and the next read. The log lives at __consumer_offsets__.log inside the broker's data directory. GetOffset returns ErrOffsetNotFoundForID (via broker/consumer/error.go) when nothing has been committed yet, which the fetch-offset handler turns into an empty response the consumer CLI treats as "start from 0."

Recovery replays the whole log on startup, exactly once:

// broker/consumer/consumer.go
func (c *ConsumerManager) Recover() error {
    c.recoverOnce.Do(func() {
        c.mu.Lock()
        defer c.mu.Unlock()
        offset := c.offsetLog.LowestOffset()
        highestOffset := c.offsetLog.HighestOffset()
        for ; offset < highestOffset; offset++ {
            data, err := c.offsetLog.Read(offset)
            if err != nil {
                c.recoverErr = err
                return
            }
            parts := strings.Split(string(data), ",")
            if len(parts) != 3 {
                continue
            }
            off, err := strconv.ParseUint(parts[2], 10, 64)
            if err != nil {
                c.recoverErr = err
                return
            }
            if _, ok := c.offsetCache[parts[0]]; !ok {
                c.offsetCache[parts[0]] = make(map[string]uint64)
            }
            c.offsetCache[parts[0]][parts[1]] = off
        }
    })
    return c.recoverErr
}

sync.Once means Recover is safe to call defensively from more than one place, but the replay itself only ever runs once per process. NewConsumerManager calls it eagerly during construction, so by the time the broker starts serving requests the cache is already warm — no per-request recovery cost. Because the log is a straight append and the replay applies entries in order, each (consumerID, topic) pair simply ends up with whatever offset was committed last; earlier commits for the same pair are silently superseded, which is exactly the semantics a commit log should have.

Step 13: Choose an ack mode when producing

Client.Send/SendBatch and ProducerClient.Produce/ProduceBatch all take a protocol.AckMode (api/protocol/types.go, covered on the wire-protocol page). It controls how long the leader waits before responding:

Mode	Behavior	Use when
AckLeader	Leader appends locally and responds — the default.	General-purpose messaging; a good balance.
AckAll	Leader appends, then waits for every in-sync replica to catch up (bounded by a timeout) before responding.	Data must survive a leader crash (financial/critical records).

AckMode also defines AckNone = 0, but the broker's produce handler only recognizes AckLeader and AckAll — anything else, AckNone included, is rejected with CodeInvalidAckMode. Every produce call must pick one of the two supported modes; the producer CLI's --acks flag defaults to 1 (AckLeader) for exactly this reason.

Summary

Component	File	Purpose
RemoteClient	client/rpc.go	One connection for admin/discovery: CreateTopic, DeleteTopic, FindTopicLeader, FindRaftLeader, ListTopics.
TryAddrs	client/bootstrap.go	Tries each bootstrap address in turn until one yields a non-empty result.
ResolveTopicLeader / ReconnectBackoff	client/leader.go	Leader lookup wrapped as a resolver function, plus linear-backoff retry (10 attempts, 500ms steps) around any connect call.
ShouldReconnect / RetryTopicNotReady	client/reconnect.go, client/retry.go	The two retry tiers above bootstrap: reconnect-worthy vs. retry-in-place-briefly.
DialWithFallback	api/transport/transport.go	Every RPC client's single dial path; retries 127.0.0.1:<port> when a Docker-internal hostname doesn't resolve.
ProducerClient	producer/client/producer.go	Single-connection Produce/ProduceBatch, wrapped in RetryTopicNotReady.
producer/client.Client	producer/client/client.go	Self-healing producer: resolves the leader, reconnects on failover, exposes Send/SendBatch.
Producer CLI	producer/cmd/main.go	connect subcommand reads stdin, calls Client.Send with the chosen AckMode.
ConsumerClient	consumer/client/consumer.go	Single-connection Fetch/FetchBatch/CommitOffset/FetchOffset; ReplicaNodeID/DisableTopicNotReadyRetry used by the broker's replication engine.
consumer/client.Client	consumer/client/client.go	Self-healing consumer: Poll (fetch-and-wait-and-reconnect), Commit, FetchCommittedOffset.
Consumer CLI	consumer/cmd/main.go	Resolves start offset (--from-beginning > --offset > last committed), loops Poll → print → Commit.
ConsumerManager	broker/consumer/consumer.go	Server-side committed-offset cache, persisted to append-only __consumer_offsets__.log, recovered once at startup via sync.Once.
AckMode	api/protocol/types.go	AckNone / AckLeader / AckAll — how long a producer waits for durability.

With client libraries and offset tracking in place, the next page shows how to build and run the entire cluster — Docker Compose setup, producing and consuming messages against a live cluster, and testing fault tolerance by killing and restarting nodes.