Acknowledged Offset Commits

Introduction to the Lesson

Today, we will make that progress durable. This lesson directly addresses the learning outcome: Update the consumer group coordinator to commit a consumer's offset only upon receiving an acknowledgment.

"Committing an offset" means creating a durable record of how far a consumer group has read into a partition. This is the final, critical step in implementing the "at-least-once" delivery guarantee, ensuring that even if consumers or brokers crash, the system can recover and resume from the correct position.

We will cover:

1. The core concepts of offset commits and how professional systems like Kafka manage them.
2. Designing a database schema in PostgreSQL to durably store these offsets.
3. Updating our consumer group coordinator to handle Ack messages by persisting the new offset.
4. Implementing the database logic in our storage repository to perform the commit.

By the end of this lesson, our broker will have a fault-tolerant mechanism for tracking consumer progress.

1. Understanding Offset Commits

Before we write any code, let's solidify our understanding of what an offset commit is and why it's so fundamental. When a consumer group "commits" an offset, it's telling the broker, "I guarantee that I have successfully processed all messages up to this point in this partition." The broker then stores this information.

To see how this works in a production system like Kafka, please watch the following video. It explains the concept of offset commits, the different strategies, and how Kafka uses a special internal topic to store them.

The key takeaway is that we are implementing at-least-once semantics. We process the message first, and only then do we commit the offset. This ensures no message is lost, though it opens the possibility of duplicates if a crash occurs after processing but before the commit succeeds.

2. Designing the Storage for Consumer Offsets

Kafka stores offsets in a special, compacted topic named __consumer_offsets. This is a clever design that reuses the broker's existing log infrastructure. We'll achieve the same goal using our PostgreSQL backend, which aligns with our project's architecture.

The central question is: what information do we need to uniquely identify a consumer group's position? As the previous video alluded to, and as we'll see confirmed in this reading, it's a combination of identifiers.

The article reveals the "recipe" for our database schema. To store an offset, we need to know:

• The consumer group (group_id)
• The topic (topic)
• The partition number (partition)

The offset is the value we store for this unique combination. This leads us to a straightforward table design.

SQL Schema for consumer_group_offsets:

CREATE TABLE consumer_group_offsets (
    group_id TEXT NOT NULL,
    topic TEXT NOT NULL,
    partition INT NOT NULL,
    current_offset BIGINT NOT NULL,
    PRIMARY KEY (group_id, topic, partition)
);

The composite PRIMARY KEY is the most important part of this design. It enforces that for any given partition of a topic, a consumer group can only have one committed offset. When we need to update the offset, we won't be inserting a new row every time. Instead, we'll use an UPSERT operation (INSERT ... ON CONFLICT), which atomically creates the row if it's missing or updates it if it exists. This elegantly mimics the "log compaction" behavior described in the article.

3. Updating the Coordinator to Handle Commits

Now, let's bring this logic into our Coordinator. In the previous lesson, we left off with a HandleAck method that just logged the incoming acknowledgment. It's time to replace that with the real implementation.

First, we need to give our Coordinator access to the storage layer. You'll likely pass your storage interface to the Coordinator's constructor.

// In your coordinator/coordinator.go file

type Coordinator struct {
    // ... existing fields like groups, mu, etc.
    storage Storage // Your storage interface
}

func NewCoordinator(storage Storage) *Coordinator {
    return &Coordinator{
        storage: storage,
        groups:  make(map[string]*ConsumerGroup),
        // ...
    }
}

Next, let's define the new method on our Storage interface.

// In your storage/storage.go (or equivalent)

type Storage interface {
    // ... other methods like Append, Read, etc.
    CommitOffset(ctx context.Context, groupID, topic string, partition int32, offset int64) error
}

Now, we can implement the HandleAck logic. The flow is:

1. Receive the Ack.
2. Find the relevant consumer group.
3. Ask the storage layer to persist the new offset.

The offset we commit is ack.Offset + 1, as this represents the next message the consumer should receive.

// In your coordinator/coordinator.go file

func (c *Coordinator) HandleAck(ctx context.Context, ack *protocol.Ack) {
    // The coordinator's primary job here is to persist the offset.
    // In more complex scenarios, you might update an in-memory cache first,
    // but direct persistence is a solid and simple starting point.

    // The committed offset is the offset of the successfully processed message + 1.
    commitOffset := ack.Offset + 1

    err := c.storage.CommitOffset(ctx, ack.GroupID, ack.Topic, ack.Partition, commitOffset)
    if err != nil {
        log.Printf(
            "Failed to commit offset for GroupID=%s, Topic=%s, Partition=%d, Offset=%d: %v",
            ack.GroupID,
            ack.Topic,
            ack.Partition,
            commitOffset,
            err,
        )
        // In a production system, you might retry or have more robust error handling.
        return
    }

    log.Printf(
        "Successfully committed offset: GroupID=%s, Topic=%s, Partition=%d, Offset=%d",
        ack.GroupID,
        ack.Topic,
        ack.Partition,
        commitOffset,
    )
}

// You will also need to update the call site in your server's handleConnection
// method to pass the context.
// s.coordinator.HandleAck(r.Context(), &ack)

4. Implementing the PostgreSQL Commit

The final step is to implement the CommitOffset method in our PostgreSQL repository. This is where we'll use the UPSERT query we designed earlier.

// In your storage/postgres.go file

func (db *PostgresStorage) CommitOffset(ctx context.Context, groupID, topic string, partition int32, offset int64) error {
    query := `
        INSERT INTO consumer_group_offsets (group_id, topic, partition, current_offset)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (group_id, topic, partition)
        DO UPDATE SET current_offset = EXCLUDED.current_offset;
    `
    
    _, err := db.pool.Exec(ctx, query, groupID, topic, int(partition), offset)
    if err != nil {
        return fmt.Errorf("failed to commit offset: %w", err)
    }
    
    return nil
}

A quick note on the SQL query:

• ON CONFLICT (group_id, topic, partition) tells PostgreSQL to use the primary key to detect a conflict.
• DO UPDATE SET current_offset = EXCLUDED.current_offset is the action to take on conflict. EXCLUDED is a special keyword that refers to the values from the row that would have been inserted if there were no conflict. This ensures the current_offset is always updated to the latest value provided.

With this implementation, every Ack message from a consumer results in a durable, atomic update to your PostgreSQL database, correctly tracking the group's progress.

Conclusion

Congratulations! You have now closed a major reliability loophole in our message broker. The system no longer just "hopes" a message was processed; it requires proof via an acknowledgment and durably records the consumer group's progress.

Key Takeaways:

• Offset Committing is the act of durably storing a consumer group's progress in a topic partition.
• We modeled our storage solution on Kafka's principles, using a database table with a (group_id, topic, partition) primary key to ensure we only store the latest offset.
• The INSERT ... ON CONFLICT (UPSERT) statement is the ideal tool for this, providing an atomic "create or update" operation.
• The Coordinator orchestrates this process, taking Ack messages from the network layer and instructing the Storage layer to persist the commit.

Next Up

Our system is now more robust against losing processed messages. But what if a consumer takes a message and then crashes or becomes unresponsive without ever sending an ack? The broker would wait forever.

In our next lesson, we will address this by implementing a redelivery mechanism for messages that are not acknowledged within a configured timeout. This will complete the core logic for at-least-once delivery by handling both successful and failed processing scenarios.