Stage 7 · Master
Phase 16 — Kafka Integration
Consumer Groups
Wire the actual Reader loop Lesson 3's Handler plugs into, understand what 6 partitions actually bounds, and prove a rebalance mid-flight never causes a silent drop.
6 Partitions Is a Ceiling on Parallelism, Not a Target to Hit
Lesson 1 fixed payment.completed at 6 partitions. That number sets a hard ceiling: at most 6 consumer instances in the notification-service consumer group can ever be doing useful work against this topic at once, because Kafka never assigns two members of the same group the same partition simultaneously. A 7th replica joins the group, sends heartbeats, and is assigned zero partitions — it is not an error, just idle capacity waiting for a partition to free up.
GroupID Is What Actually Makes This a Consumer Group
package consumer
import (
"context"
kafkago "github.com/segmentio/kafka-go"
)
// NewReader disables auto-commit (CommitInterval: 0) deliberately. Auto-commit
// would advance the offset on a fixed timer regardless of whether this
// process's own database transaction had actually committed yet — a crash
// in that window would silently skip the message forever instead of
// redelivering it.
func NewReader(brokers []string) *kafkago.Reader {
return kafkago.NewReader(kafkago.ReaderConfig{
Brokers: brokers,
Topic: "payment.completed",
GroupID: consumerGroup,
GroupBalancers: []kafkago.GroupBalancer{kafkago.RangeGroupBalancer{}},
CommitInterval: 0,
MinBytes: 1,
MaxBytes: 1 << 20,
})
}
// Run fetches one message at a time and only commits its offset after
// handler has already committed its own transaction — a crash between
// those two lines redelivers the message rather than dropping it.
func Run(ctx context.Context, reader *kafkago.Reader, handler *Handler) error {
for {
msg, err := reader.FetchMessage(ctx)
if err != nil {
return err
}
if err := handler.HandlePaymentCompleted(ctx, msg.Topic, msg.Value); err != nil {
return err // Lesson 5 turns this into bounded retry instead of returning immediately
}
if err := reader.CommitMessages(ctx, msg); err != nil {
return err
}
}
}
package main
import (
"context"
"os"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/rs/zerolog/log"
"github.com/hoa-platform/backend/services/notification-service/internal/consumer"
)
func main() {
ctx := context.Background()
pool, err := pgxpool.New(ctx, os.Getenv("NOTIFICATION_SERVICE_DB_URL"))
if err != nil {
log.Error().Err(err).Msg("consumer: connect to database")
os.Exit(1)
}
defer pool.Close()
reader := consumer.NewReader([]string{"localhost:9092"})
defer reader.Close()
handler := consumer.NewHandler(pool, consumer.NewInboxStore())
if err := consumer.Run(ctx, reader, handler); err != nil {
log.Error().Err(err).Msg("consumer: run loop exited")
os.Exit(1)
}
}
Starting a Second Replica Triggers a Visible Rebalance
Starting a second process with the same GroupID ("notification-service") makes Kafka trigger a rebalance: RangeGroupBalancer reassigns contiguous partition ranges across however many group members are currently alive. During that reassignment, FetchMessage on affected readers blocks briefly. If the first replica had fetched a message but not yet committed its offset when the rebalance moved that partition to the second replica, the second replica's next fetch on that partition starts from the last committed offset — meaning it may reprocess a message the first replica was mid-handling. That reprocessing is safe for the same reason a producer-crash duplicate is safe: Lesson 3's inbox claim rejects it.
Every organization's payment.completed events flow through the same single consumer group and the same 6 partitions — there is no per-tenant consumer group, and there never should be. Isolation between organizations is enforced the same way it is everywhere else on this platform: by the organization_id column on notification_jobs and the org_id field inside every event, not by giving each tenant its own infrastructure to reason about.
Watch kafka-consumer-groups.sh Split Partitions Across Two Replicas
Applied exercise
Prove per-payment ordering survives a rebalance in flight
This lesson asserts a rebalance mid-processing is safe, not just fast. Prove it by forcing one.
- Publish 3 events with the same aggregate_id (so they share a partition) but distinct event_id values, spaced roughly 1 second apart.
- While the first replica is processing the second of the three, start a second consumer replica to force a rebalance mid-stream.
- After both events finish processing, query notification_jobs and confirm exactly 3 rows exist (not fewer, from a dropped message, and not more, from a duplicate).
- Explain in one paragraph why the third event, which may be handled by either replica depending on how the rebalance lands, still ends up correctly recorded exactly once.
Deliverable
A short paragraph plus the row count from notification_jobs after the exercise.
Completion checks
- Exactly 3 notification_jobs rows exist for the 3 published events, regardless of which replica handled each one.
- The explanation connects the correct outcome to the same inbox-claim mechanism from Lesson 3, not to luck or timing.
With 6 partitions on payment.completed and 8 running replicas in the notification-service consumer group, what happens to the 2 extra replicas?
Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.