Message Acknowledgment in Messaging Systems

Introduction to the Lesson

Today, we're closing that gap. This lesson's learning outcome is to implement a consumer acknowledgment mechanism where consumers confirm message processing. This is a fundamental feature for building reliable systems, as it allows the broker to track a consumer's progress and forms the basis for "at-least-once" delivery semantics.

We will cover:

1. The concept of delivery guarantees and the role of manual acknowledgments.
2. Extending our binary protocol to support a new Ack message.
3. Modifying the consumer to send an acknowledgment after it successfully processes a message.
4. Updating the broker to receive and handle these acknowledgments.

Let's begin by understanding why this is so critical.

1. Delivery Guarantees and the Problem of Lost Work

When a consumer fetches a message at offset N, the naive approach is to assume it's done and let the consumer ask for offset N+1 next time. But what happens if the consumer process crashes before it can write the result to its database? The work is lost, but the offset has moved on.

This is where delivery guarantees come in. By requiring the consumer to send a confirmation—an acknowledgment or "ack"—after it has completed its work, we can build a more robust system.

The blog post from OneUptime provides an excellent overview of acknowledgment strategies.

As the article highlights, manual acknowledgment is the foundation for reliable processing. We are shifting the responsibility of confirming completion from the broker to the consumer itself. This ensures that we can achieve an "at-least-once" delivery guarantee: a message might be delivered more than once (e.g., if the consumer processes it but crashes before sending the ack), but it will never be silently lost.

2. Extending the Broker Protocol for Acknowledgments

To implement this, our consumer needs to send a new type of message to the broker. This requires extending the binary protocol we designed in Module 5.

First, let's add a new message type constant.

// In your protocol or constants package

const (
    // ... existing message types
    MsgTypeJoinGroup  = 5
    MsgTypeLeaveGroup = 6
    MsgTypeSyncGroup  = 7
    MsgTypeAck        = 8 // New message type for acknowledgments
)

Next, we define the structure for our Ack message. What information does the broker need to identify which message is being acknowledged?

• The GroupID of the consumer.
• The Topic and Partition the message came from.
• The Offset of the message that was successfully processed.

Here is the Go struct for our acknowledgment message.

// In your protocol package, alongside other message structs

type Ack struct {
    Topic      string
    Partition  int32
    GroupID    string
    ConsumerID string // The specific consumer sending the ack
    Offset     int64
}

// You will also need to implement the Encode and Decode methods for this struct,
// similar to how you did for ProduceRequest, ConsumeRequest, etc.
// This typically involves using binary.Write and binary.Read.

With this in place, both our client and server can be updated to understand this new message type.

3. Implementing the Consumer-Side Acknowledgment

The consumer's responsibility is simple: after it receives and successfully processes a message, it must send an Ack back to the broker.

Let's modify the consumer's main processing loop. Previously, this loop might have just printed the message. Now, it will perform its work and then trigger the acknowledgment.

// In your consumer client logic

// processMessages is the main loop for a consumer assigned to a partition.
func (c *Consumer) processMessages(ctx context.Context, topic string, partition int32) {
    // ... logic to connect to the broker ...

    for {
        // 1. Receive a message from the broker (your existing logic)
        msg, err := c.fetchMessage(ctx, topic, partition, c.currentOffset)
        if err != nil {
            log.Printf("Failed to fetch message: %v", err)
            // Add retry/backoff logic here
            continue
        }

        // 2. Process the message. This is your business logic.
        err = c.handler(msg.Value) // handler is a user-provided function
        if err != nil {
            log.Printf("Failed to process message at offset %d: %v", msg.Offset, err)
            // Decide on a retry/DLQ strategy. For now, we'll stop.
            // In a real system, you would not ack here.
            continue
        }

        // 3. On success, send the acknowledgment.
        err = c.sendAck(ctx, topic, partition, msg.Offset)
        if err != nil {
            log.Printf("Failed to send ack for offset %d: %v", msg.Offset, err)
            // If ack fails, the message will be reprocessed later,
            // which is the correct "at-least-once" behavior.
        }

        // 4. Update the local offset for the next fetch.
        c.currentOffset = msg.Offset + 1
    }
}

// sendAck constructs and sends an Ack message to the broker.
func (c *Consumer) sendAck(ctx context.Context, topic string, partition int32, offset int64) error {
    ackMsg := &protocol.Ack{
        Topic:      topic,
        Partition:  partition,
        GroupID:    c.groupID,
        ConsumerID: c.consumerID,
        Offset:     offset,
    }

    // You will need a function to encode and send this message
    // over the TCP connection, prepended with the MsgTypeAck byte.
    return c.conn.Send(protocol.MsgTypeAck, ackMsg)
}

This flow ensures that the acknowledgment is only sent after the business logic (c.handler) completes without an error.

4. Implementing the Broker-Side Handler

Now that the consumer is sending Ack messages, the broker needs to be able to receive and act on them. We'll update the server's connection handler—the goroutine that manages communication with a single client.

In the loop where you read the message type byte and then decode the rest of the message, you will add a new case.

// In your broker's server connection handler

func (s *Server) handleConnection(conn net.Conn) {
    // ... connection setup ...
    for {
        // Read message type (1 byte)
        msgType, err := reader.ReadByte()
        // ... error handling ...

        switch msgType {
        // ... case for MsgTypeProduce, MsgTypeConsume, etc.

        case protocol.MsgTypeAck:
            var ack protocol.Ack
            // Decode the Ack message from the connection
            if err := ack.Decode(reader); err != nil {
                log.Printf("Failed to decode ack message: %v", err)
                continue // Or close connection
            }
            
            // The broker has received the acknowledgment.
            // Now, it must inform the coordinator to record this progress.
            s.coordinator.HandleAck(&ack)
        }
    }
}

The final step for today is to define the HandleAck method on our Coordinator. For now, its job will be to simply receive the information. The actual persistence of this offset is a critical step that we will tackle in the next lesson.

// In your coordinator implementation

func (c *Coordinator) HandleAck(ack *protocol.Ack) {
    // For now, we will just log that we received the acknowledgment.
    // In the next lesson, we will use this information to update the
    // committed offset for the consumer group.
    log.Printf(
        "Received ack: GroupID=%s, Topic=%s, Partition=%d, Offset=%d",
        ack.GroupID,
        ack.Topic,
        ack.Partition,
        ack.Offset,
    )

    // Future logic:
    // 1. Lock the consumer group state.
    // 2. Find the group and partition.
    // 3. Update the committed offset.
    // 4. Persist the offset to storage (e.g., PostgreSQL).
    // 5. Unlock.
}

By receiving this Ack, the broker is now aware of the consumer's progress. This completes the communication loop for our acknowledgment mechanism.

Conclusion

In this lesson, you have significantly improved the reliability of our message broker. By moving from a "fire-and-forget" model to one of explicit confirmation, you've laid the groundwork for durable and predictable message processing.

Key Takeaways:

• Manual Acknowledgment is crucial for implementing "at-least-once" delivery semantics, preventing data loss if a consumer fails mid-process.
• The acknowledgment flow required extending our binary protocol with a new Ack message type and corresponding data structure.
• The consumer's role is to process a message fully before sending the acknowledgment for that message's offset.
• The broker's role is to receive these acknowledgments and pass them to the coordinator, which is responsible for tracking the group's progress.

Next Up

Our broker now receives acknowledgments but doesn't do anything durable with them yet. If the broker restarts, it will forget all progress. In the next lesson, we will complete the feature by implementing the learning outcome: Update the consumer group coordinator to commit a consumer's offset only upon receiving an acknowledgment. This will involve persisting the offset to our PostgreSQL backend, making the consumer group's progress fault-tolerant.