When a service receives a stop signal (SIGTERM), there is a risk of losing messages: the consumer may have received an event from Kafka but not managed to process it and commit the offset — after a restart, Kafka will hand out this message again. The producer may lose a few messages that have not yet gone out to the broker.
In Spring Boot these details are hidden inside ConcurrentMessageListenerContainer: the container itself drains the current batch and commits the offset. In Go the same work must be written by hand — it is no harder, but it requires understanding the correct order of actions.
How consumer shutdown works
The consumer in kafka-go reads messages in an infinite loop via FetchMessage. To stop this loop cleanly, context.Context is used: when the context is cancelled on SIGTERM, FetchMessage returns a context.Canceled error — and this is a normal exit, not a failure.
An important detail: kafka-go has no automatic offset commit. The offset must be committed explicitly by calling CommitMessages — after each successfully processed message.
func (c *OrderConsumer) Run(ctx context.Context) error {
for {
msg, err := c.reader.FetchMessage(ctx)
if err != nil {
if errors.Is(err, context.Canceled) || errors.Is(err, io.EOF) {
return nil // normal exit on shutdown
}
return fmt.Errorf("fetch message: %w", err)
}
if err := c.handle(ctx, msg); err != nil {
return fmt.Errorf("handle order event: %w", err)
}
if err := c.reader.CommitMessages(ctx, msg); err != nil {
return fmt.Errorf("commit offset: %w", err)
}
}
}
When the context is cancelled, FetchMessage will finish only after the current message has already been processed and the offset committed. No replay occurs.
The context.Canceled and io.EOF errors are a normal consumer exit on shutdown; they should not be logged as errors. Otherwise the alert channel will be noisy on every deploy.
Registering the goroutine in a WaitGroup
So that the main process waits for the consumer to finish before closing the connection pool, the goroutine is registered in a sync.WaitGroup:
consumerCtx, cancelConsumer := context.WithCancel(ctx)
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
if err := orderConsumer.Run(consumerCtx); err != nil {
slog.ErrorContext(ctx, "order consumer exited with error", "error", err)
}
}()
wg.Add(1) comes before starting the goroutine — so there is no race between the goroutine start and the wg.Wait() call in shutdown.
What to do if a message arrives again
If SIGTERM arrived after a successful handle but before CommitMessages, then on the next service run Kafka will hand out this message once more. Therefore the handler must be idempotent — reprocessing the same event must not lead to a double effect.
The standard technique is a table of already-processed events:
func (c *OrderConsumer) handle(ctx context.Context, msg kafka.Message) error {
var event OrderConfirmedEvent
if err := json.Unmarshal(msg.Value, &event); err != nil {
return fmt.Errorf("unmarshal order event: %w", err)
}
tx, err := c.pool.Begin(ctx)
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback(ctx)
q := c.queries.WithTx(tx)
inserted, err := q.InsertProcessedEvent(ctx, db.InsertProcessedEventParams{
EventID: event.ID,
ConsumerGroup: "billing-confirmations",
})
if err != nil {
return fmt.Errorf("dedup check: %w", err)
}
if !inserted {
return nil // already processed earlier
}
if err := c.orders.RecordConfirmation(ctx, event.OrderID, event.TotalAmount); err != nil {
return fmt.Errorf("record confirmation: %w", err)
}
return tx.Commit(ctx)
}
InsertProcessedEvent uses ON CONFLICT (event_id, consumer_group) DO UPDATE ... RETURNING (xmax = 0) AS inserted. On the first insert it returns true, on a retry — false. It is important that the duplicate check and the business logic itself are in one transaction: if RecordConfirmation fails, the transaction rolls back and the event will be reprocessed on the next run.
Why you cannot make a long HTTP request inside handle
A common mistake is to call an external HTTP service from the handler with several retries:
// A dangerous variant — do not do this
func (c *ProductConsumer) handle(ctx context.Context, msg kafka.Message) error {
var event ProductPriceChangedEvent
_ = json.Unmarshal(msg.Value, &event)
// HTTP + retry → can take 20-30 seconds
if err := c.catalogClient.UpdatePrice(ctx, event.ProductID, event.NewPrice); err != nil {
return fmt.Errorf("update price: %w", err)
}
return nil
}
On SIGTERM the context is cancelled and the HTTP client gets context.Canceled. The processing ends up partial, and the shutdown hangs for the duration of the timeout.
The correct solution is to write only to the database and put the event into an outbox table. A separate relay goroutine will pick it up and send the HTTP independently:
func (c *ProductConsumer) handle(ctx context.Context, msg kafka.Message) error {
var event ProductPriceChangedEvent
if err := json.Unmarshal(msg.Value, &event); err != nil {
return fmt.Errorf("unmarshal product event: %w", err)
}
return sqlcTx(ctx, c.pool, func(q *db.Queries) error {
if err := q.UpdateProductPrice(ctx, db.UpdateProductPriceParams{
ProductID: event.ProductID,
Price: event.NewPrice,
}); err != nil {
return fmt.Errorf("update product price: %w", err)
}
return q.InsertOutboxEvent(ctx, db.InsertOutboxEventParams{
EventType: "ProductPriceChanged",
Payload: msg.Value,
})
})
}
This way handle finishes in a few milliseconds, and the relay works on its own budget.
Stopping the producer: why you need writer.Close()
kafka.Writer accumulates messages into a batch and sends them to the broker asynchronously. If the process exits without an explicit Close() call, the messages from the current unfinished batch are simply lost — they will not reach Kafka.
type OrderPublisher struct {
writer *kafka.Writer
}
func NewOrderPublisher(brokers []string) *OrderPublisher {
return &OrderPublisher{
writer: &kafka.Writer{
Addr: kafka.TCP(brokers...),
Topic: "orders.confirmed",
Balancer: &kafka.LeastBytes{},
BatchTimeout: 5 * time.Millisecond,
},
}
}
func (p *OrderPublisher) Publish(ctx context.Context, event OrderConfirmedEvent) error {
payload, err := json.Marshal(event)
if err != nil {
return fmt.Errorf("marshal order confirmed: %w", err)
}
return p.writer.WriteMessages(ctx, kafka.Message{
Key: []byte(event.OrderID),
Value: payload,
})
}
func (p *OrderPublisher) Close() error {
return p.writer.Close()
}
writer.Close() blocks until all accumulated messages have been sent to the broker. Only after that does it return — and only then can you close the remaining resources.
The correct shutdown order
The order in which components are stopped matters. You cannot close the connection pool before the consumer has finished CommitMessages — pgx will panic when trying to acquire a connection from a closed pool.
shutdownFns := []func(){
func() { appState.SetNotReady() }, // stop accepting traffic
func() { cancelConsumer() }, // signal the consumer to stop
func() { wg.Wait() }, // wait for CommitMessages to finish
func() {
if err := orderPublisher.Close(); err != nil {
slog.ErrorContext(ctx, "kafka writer close", "error", err)
}
}, // flush the producer's pending batch
func() { srv.Shutdown(shutCtx) }, // HTTP: wait for the current requests
func() { pool.Close() }, // DB — the very last
}
The logic is simple: first stop what accepts new work, then wait for the current work to finish, and only then close the shared resources (HTTP, DB).
Common mistakes
Auto-commit via CommitInterval. In kafka-go there is a CommitInterval option on kafka.Reader that automatically commits the offset at a given interval. It is convenient but dangerous: the offset can be committed before the message is actually processed. On SIGTERM, some messages will be considered processed even though the handler never reached them. You need an explicit CommitMessages after each successful handle.
Starting a goroutine without a WaitGroup. If you do not wait for the consumer to finish before pool.Close(), the goroutine may try to acquire a connection from an already-closed pool.
Skipping writer.Close(). Without explicitly closing the writer, the last few messages will remain in the buffer and will not reach Kafka.
Logging context.Canceled as an error. This is a normal way for the consumer to finish — it should not be included in alerts.
In short
- The consumer is stopped by cancelling
context.Context—FetchMessagereturnscontext.Canceled, and this is a normal exit. - kafka-go has no auto-commit: the offset must be committed explicitly via
CommitMessagesafter each successfulhandle. - The consumer goroutine is registered in a
sync.WaitGroup— shutdown waits for it viawg.Wait()before closing the pool. - If SIGTERM arrived between
handleandCommitMessages, the message will arrive again — the handler must be idempotent. - Long HTTP requests with retries from
handleare dangerous: on context cancellation the processing will be partial. Use the outbox pattern. writer.Close()blocks and flushes the accumulated batch to the broker — without it, the last messages are lost.- The shutdown order: cancel the context →
wg.Wait()→writer.Close()→ HTTP drain →pool.Close().
What to read next
- HTTP drain in Go —
srv.Shutdown(ctx), preStop sleep, long endpoints. - DB and persistence in Go — the order of
pool.Close(), transactions in background goroutines. - Scheduled / Async / outbox in Go — the outbox-relay,
ctx.Done()before a new iteration. - Budgets and observability in Go — the cumulative 60s, the
app_shutdown_duration_secondsmetric.