Messaging Kafka: Producer dan Consumer

Event berpindah antar proses lewat Kafka. Sisi tulis berada di paket internal/gateway/messaging dan dipanggil use case setelah transaksi database di-commit. Sisi baca berada di paket internal/delivery/messaging dan dijalankan oleh cmd/worker. Kedua sisi memakai Sarama. Kontrak event sangat tipis: satu interface Event dengan method GetId, internal/model/event.go:3-5¹.

Producer generik

Struct Producer[T model.Event] menyimpan tiga hal saja: sarama.SyncProducer, nama Topic, dan logrus.Logger untuk logging, internal/gateway/messaging/producer.go:11-15².

Method Send melakukan json.Marshal atas event. Kegagalan marshal dicatat sebagai failed to marshal event lalu dikembalikan. Pesan disusun sebagai sarama.ProducerMessage dengan Key berisi sarama.StringEncoder(event.GetId()) dan Value berisi byte JSON, sehingga partisi ditentukan oleh ID event, internal/gateway/messaging/producer.go:21-32³.

Karena producer bertipe sinkron, SendMessage mengembalikan partition dan offset. Nilai itu dicatat lewat Log.Debugf dengan format Message sent to topic %s, partition %d, offset %d. Kegagalan kirim dicatat sebagai failed to produce message, internal/gateway/messaging/producer.go:34-41⁴.

graph LR
  UC[Usecase] --> PR[Producer Send]
  CFG[NewKafkaProducer] --> PR
  PR --> MSG[sarama.ProducerMessage]
  MSG --> K[Kafka brokers]
  K --> T1[users]
  K --> T2[contacts]
  K --> T3[addresses]
  T1 --> CG[ConsumeTopic]
  T2 --> CG
  T3 --> CG
  CG --> CC[Consumer Consume]
  style K fill:#336791,color:#fff
  style PR fill:#f59e0b,color:#000
  style CFG fill:#f59e0b,color:#000
  style CG fill:#8b5cf6,color:#fff
  style CC fill:#2d7d46,color:#fff
  style UC fill:#2d7d46,color:#fff
%% fig: Jalur satu event dari use case ke consumer
%% desc: Use case memanggil Producer Send, Sarama mengirim pesan ke broker, lalu ConsumeTopic mengambilnya dari topic yang sama.

Satu producer per topik

Tiga tipe membungkus producer generik dan mengunci nama topik di dalam constructor-nya. Ketiganya menerima sarama.SyncProducer dan logger.

Membuat producer sinkron dari konfigurasi

NewKafkaProducer lebih dulu membaca kafka.producer.enabled. Bila false, fungsi mencatat Kafka producer is disabled dan mengembalikan nil. Bila aktif, konfigurasi Sarama diisi Producer.Return.Successes = true, Producer.RequiredAcks = sarama.WaitForAll, dan Producer.Retry.Max = 3. Daftar broker diambil dari kafka.bootstrap.servers yang dipisah koma, lalu sarama.NewSyncProducer dipanggil; kegagalan berakhir di log.Fatalf("Failed to create producer: %v", err), internal/config/kafka.go:32-50⁸.

Nilai bawaan di config.json justru mematikan producer. Blok kafka memuat bootstrap.servers bernilai localhost:9092, group.id bernilai golang-clean-architecture, auto.offset.reset bernilai earliest, dan producer.enabled bernilai false, config.json:24-39⁹.

Karena producer bisa nil, Bootstrap mendeklarasikan ketiga producer lebih dulu lalu hanya membuatnya di dalam blok if config.Producer != nil. Hasilnya disuntikkan ke NewUserUseCase, NewContactUseCase, dan NewAddressUseCase, internal/config/app.go:34-43¹⁰.

Use case juga memeriksa nil sebelum menerbitkan event. Pada ContactUseCase.Create, event dibentuk lewat converter.ContactToEvent(contact) setelah tx.Commit() berhasil; bila producer nil, log menulis Kafka producer is disabled, skipping contact created event. Bila Send gagal, use case mencatat error publishing contact created event dan mengembalikan fiber.ErrInternalServerError, internal/usecase/contact_usecase.go:65-74¹¹.

⚠️ Dengan producer.enabled bernilai false, tidak ada event yang pernah terbit dan consumer tidak menerima apa pun. Selain itu, kegagalan Send terjadi setelah commit, sehingga data sudah tersimpan tetapi response bernilai 500 dan event untuk perubahan itu hilang.

Consumer group dari konfigurasi

NewKafkaConsumerGroup mengisi Consumer.Return.Errors = true agar error group bisa dibaca. Offset awal ditentukan dari kafka.auto.offset.reset: nilai earliest memilih sarama.OffsetOldest, nilai lain memilih sarama.OffsetNewest. Broker dipisah dari kafka.bootstrap.servers dan groupID diambil dari kafka.group.id, lalu sarama.NewConsumerGroup dipanggil; kegagalannya berhenti di log.Fatalf("Failed to create consumer group: %v", err), internal/config/kafka.go:11-30¹².

Helper ConsumeTopic

Ada dua tipe pendukung: ConsumerHandler, yaitu func(message *sarama.ConsumerMessage) error, dan ConsumerGroupHandler yang menyimpan handler tersebut beserta logger, internal/delivery/messaging/consumer.go:10-15¹³.

ConsumeTopic membungkus handler menjadi ConsumerGroupHandler, lalu menjalankan goroutine yang memanggil consumerGroup.Consume(ctx, []string{topic}, consumerHandler) di dalam loop. Loop berhenti saat ctx.Err() != nil dengan log Context cancelled, stopping consumer, internal/delivery/messaging/consumer.go:46-63¹⁴.

Goroutine kedua membaca consumerGroup.Errors() dan mencatat setiap error sebagai Consumer group error. Setelah itu ConsumeTopic menunggu <-ctx.Done(), mencatat Closing consumer group for topic: %s, dan memanggil consumerGroup.Close(); kegagalan close dicatat sebagai Error closing consumer group, internal/delivery/messaging/consumer.go:65-76¹⁵.

Method Setup dan Cleanup pada ConsumerGroupHandler sengaja kosong dan langsung mengembalikan nil. Seluruh pekerjaan ada di ConsumeClaim, yang membaca claim.Messages() dan memanggil h.Handler(message). Bila handler mengembalikan error, hanya muncul log Failed to process message; bila sukses, pesan ditandai dengan session.MarkMessage(message, ""). Loop berhenti ketika session.Context().Done() tertutup, internal/delivery/messaging/consumer.go:25-44¹⁶.

⚠️ ConsumeClaim tidak mengembalikan error ke Sarama saat handler gagal. Pesan yang gagal hanya tidak ditandai, lalu loop berlanjut ke pesan berikutnya pada claim yang sama.

Consumer per topik

Setiap consumer hanya melakukan unmarshal JSON menjadi event dan belum memproses isinya; ketiganya masih menyisakan komentar // TODO process event dan mencatat pesan masuk berikut partition asalnya.

Kegagalan unmarshal pada masing-masing consumer dicatat dengan awalan error unmarshalling diikuti nama event, lalu error dikembalikan sehingga pesan tidak ditandai, internal/delivery/messaging/contact_consumer.go:21-31¹⁸.

Worker dan graceful shutdown

main pada worker membuat ctx dan cancel, lalu menjalankan tiga goroutine sekaligus: RunUserConsumer, RunContactConsumer, dan RunAddressConsumer, cmd/worker/main.go:21-25²⁰.

Shutdown ditangani dengan signal.Notify atas syscall.SIGINT, syscall.SIGKILL, dan syscall.SIGTERM. Loop for !stop menunggu satu sinyal, mencatat Got one of stop signals, shutting down worker gracefully, SIGNAL NAME :, memanggil cancel(), lalu keluar dari loop. Setelah itu main menunggu time.Sleep(5 * time.Second) yang diberi komentar wait for all consumers to finish processing, cmd/worker/main.go:27-40²¹.

Setiap fungsi Run*Consumer membuat consumer group sendiri lewat config.NewKafkaConsumerGroup, membuat handler lewat constructor consumer, lalu memanggil messaging.ConsumeTopic dengan topic yang sesuai. RunAddressConsumer memakai topic addresses, RunContactConsumer memakai contacts, dan RunUserConsumer memakai users, cmd/worker/main.go:43-62²².

sequenceDiagram
    participant UC as ContactUseCase
    participant PR as ContactProducer
    participant K as Kafka
    participant CT as ConsumeTopic
    participant CG as ContactConsumer
    UC->>PR: Send event
    PR->>PR: json.Marshal event
    PR->>K: SendMessage topic contacts
    K-->>PR: partition dan offset
    PR-->>UC: nil
    K->>CT: ConsumeClaim message
    CT->>CG: Handler message
    CG-->>CT: nil
    CT->>CT: session.MarkMessage
    UC->>CT: cancel ctx dari sinyal berhenti
    CT->>K: consumerGroup.Close
%% fig: Siklus satu event dari publish sampai worker berhenti
%% desc: Use case mengirim event, consumer group mengambilnya, menandai pesan sukses, lalu menutup group saat context dibatalkan.

Saat sinyal diterima, cancel() membuat <-ctx.Done() pada ConsumeTopic selesai, sehingga consumer group ditutup dan loop consume berhenti. Jeda 5 detik di akhir main memberi waktu bagi proses yang masih berjalan sebelum worker keluar, internal/delivery/messaging/consumer.go:65-76¹⁵.

⚠️ syscall.SIGKILL tetap terdaftar di signal.Notify meskipun sinyal itu tidak bisa ditangkap, dan jeda time.Sleep(5 * time.Second) adalah waktu tetap, bukan penantian atas selesainya setiap consumer, cmd/worker/main.go:27-40²¹.

Topik konfigurasi, bootstrap, dan penyusunan dependensi secara umum dibahas di Konfigurasi dan Bootstrap. Peran producer sebagai bagian dari use case dibahas di Lapisan Usecase, Model, dan Converter.