Alur Event Kafka

Event Kafka dikirim oleh lapisan usecase setelah transaksi database di-commit, bukan di dalamnya. Setiap use case memeriksa apakah producer tersedia: jika c.UserProducer != nil maka event hasil converter.UserToEvent dikirim lewat Send, dan jika tidak, jalur itu hanya mencatat "Kafka producer is disabled, skipping user created event" internal/usecase/user_usecase.go:100-114¹. Producer sendiri adalah struct generik yang menyimpan sarama.SyncProducer, nama Topic, dan Log internal/gateway/messaging/producer.go:11-15².

Producer generik dan bentuk event

Producer[T].Send melakukan tiga hal secara berurutan: json.Marshal(event), menyusun sarama.ProducerMessage dengan Key berisi event.GetId(), lalu memanggil p.Producer.SendMessage(message). Kegagalan marshal dicatat sebagai "failed to marshal event" dan kegagalan kirim sebagai "failed to produce message"; bila sukses, hasilnya masuk ke log debug "Message sent to topic %s, partition %d, offset %d" internal/gateway/messaging/producer.go:21-41³. Karena itu kontrak minimum sebuah event adalah satu method GetId() string internal/model/event.go:3-5⁴, dan model.UserEvent memenuhi kontrak itu dengan mengembalikan ID internal/model/user_event.go:3-12⁵. Method GetTopic hanya mengembalikan pointer ke p.Topic dan tidak dipakai di jalur Send internal/gateway/messaging/producer.go:17-19⁶.

Nama topik tidak ditentukan oleh use case, melainkan oleh konstruktor producer per aggregate: UserProducer mengisi Topic: "users" internal/gateway/messaging/user_producer.go:10-22⁷, ContactProducer mengisi Topic: "contacts" internal/gateway/messaging/contact_producer.go:10-22⁸, dan AddressProducer mengisi Topic: "addresses" internal/gateway/messaging/address_producer.go:10-22⁹.

Siapa yang mengirim event

Hanya sebagian operasi use case yang memancarkan event. Tabel berikut memetakan operasi ke converter dan topiknya.

ContactUseCase.Delete berhenti setelah tx.Commit() dan mengembalikan nil tanpa memanggil producer, sehingga penghapusan contact tidak pernah sampai ke Kafka internal/usecase/contact_usecase.go:146-172¹⁵.

Rangkaian producer, consumer group, dan topik

graph LR
  UC["UserUseCase / ContactUseCase / AddressUseCase"] -->|"tx.Commit() lalu Send"| P["Producer[T].Send"]
  P -->|"json.Marshal, Key = GetId()"| K["topic users / contacts / addresses"]
  K --> CG["ConsumeTopic consumer group"]
  CG --> H["UserConsumer / ContactConsumer / AddressConsumer"]
  H -->|"handler nil"| M["session.MarkMessage"]
  H -->|"handler error"| E["log Failed to process message"]
  style UC fill:#f59e0b,color:#000
  style P fill:#f59e0b,color:#000
  style K fill:#336791,color:#fff
  style CG fill:#8b5cf6,color:#fff
  style H fill:#2d7d46,color:#fff
  style M fill:#2d7d46,color:#fff
  style E fill:#dc2626,color:#fff
%% fig: Jalur event dari usecase ke consumer
%% desc: Use case mengirim event setelah commit ke topik users, contacts, atau addresses; consumer group mengambilnya dan handler memutuskan MarkMessage atau log error.

Siklus satu event di sisi consumer

ConsumeTopic membungkus handler menjadi ConsumerGroupHandler, lalu menjalankan consumerGroup.Consume(ctx, []string{topic}, consumerHandler) di dalam goroutine. Setelah Consume kembali, goroutine berhenti bila ctx.Err() != nil dengan log "Context cancelled, stopping consumer". Goroutine kedua menghabiskan consumerGroup.Errors(), dan fungsi induk menunggu <-ctx.Done(), mencatat "Closing consumer group for topic: %s", lalu memanggil consumerGroup.Close() internal/delivery/messaging/consumer.go:46-76¹⁸.

sequenceDiagram
  participant UC as UseCase
  participant PR as Producer
  participant BR as Kafka broker
  participant CG as ConsumerGroupHandler
  participant HD as UserConsumer
  UC->>PR: Send(event)
  PR->>PR: json.Marshal(event)
  PR->>BR: SendMessage(topic, Key=GetId(), Value)
  BR-->>PR: partition, offset
  PR-->>UC: nil atau error
  BR->>CG: ConsumeClaim message
  CG->>HD: Handler(message)
  alt handler mengembalikan error
    CG->>CG: log "Failed to process message"
  else handler sukses
    CG->>BR: session.MarkMessage(message, "")
  end
%% fig: Siklus satu event dari Send sampai MarkMessage
%% desc: Producer mengirim event ke broker, consumer group mengambil message di ConsumeClaim, dan offset baru ditandai setelah handler sukses.

ConsumeClaim berjalan dalam for dengan select atas claim.Messages() dan session.Context().Done(). Message nil membuat fungsi return nil. Handler dipanggil lewat h.Handler(message): bila error, hanya log "Failed to process message" yang dicatat; bila sukses, session.MarkMessage(message, "") dipanggil internal/delivery/messaging/consumer.go:25-44¹⁹. Karena itu offset baru ditandai setelah pemrosesan berhasil, sehingga pengiriman bersifat at-least-once.

Handler konkretnya masih berupa kerangka. UserConsumer.Consume melakukan json.Unmarshal ke new(model.UserEvent) dengan pesan error "error unmarshalling User event", lalu mencatat "Received topic users with event: %v from partition %d" dan mengembalikan nil; blok // TODO process event menandakan pemrosesan bisnis belum diisi internal/delivery/messaging/user_consumer.go:21-31²⁰. AddressConsumer mengikuti pola yang sama untuk topic addresses internal/delivery/messaging/address_consumer.go:21-31²¹.

Topic, handler, dan worker

Setiap topic memiliki consumer group sendiri yang dijalankan sebagai goroutine terpisah di service worker.

main pada worker membuat context.WithCancel(context.Background()), menjalankan ketiga fungsi consumer sebagai goroutine, lalu menunggu SIGINT, SIGKILL, atau SIGTERM. Saat sinyal diterima, cancel() dipanggil dan proses tidur 5 detik dengan komentar "wait for all consumers to finish processing" sebelum keluar cmd/worker/main.go:16-41²⁵.

Konfigurasi yang memengaruhi alur

NewKafkaProducer mengembalikan nil dan mencatat "Kafka producer is disabled" bila kafka.producer.enabled bukan true. Jika aktif, producer dibuat dengan Producer.Return.Successes = true, Producer.RequiredAcks = sarama.WaitForAll, dan Producer.Retry.Max = 3 internal/config/kafka.go:32-50²⁶. Nil itulah yang mengalir ke Bootstrap, yang hanya membuat ketiga producer ketika config.Producer != nil internal/config/app.go:34-48²⁷, sehingga use case menerima producer nil dan melewati pengiriman event.

Di sisi consumer, NewKafkaConsumerGroup mengaktifkan Consumer.Return.Errors, memilih sarama.OffsetOldest bila kafka.auto.offset.reset bernilai earliest dan sarama.OffsetNewest bila tidak, lalu membuat group dengan kafka.group.id internal/config/kafka.go:11-30²⁸.

⚠️ Event dikirim setelah tx.Commit() sukses. Bila Send gagal, use case mengembalikan fiber.ErrInternalServerError padahal data sudah tersimpan, dan tidak ada mekanisme kompensasi di jalur itu internal/usecase/user_usecase.go:100-114¹.

⚠️ RequiredAcks = sarama.WaitForAll dengan Retry.Max = 3 dapat menghasilkan message ganda saat retry, sementara MarkMessage hanya dipanggil setelah handler sukses. Pasangkan konsumen dengan operasi idempoten sebelum mengisi blok // TODO process event internal/config/kafka.go:32-50²⁶ internal/delivery/messaging/consumer.go:25-44¹⁹.

Untuk urutan pemanggilan dari HTTP sampai use case yang memicu event, lihat Alur Request HTTP; untuk detail kelas producer dan consumer beserta konfigurasi lengkapnya, lihat Messaging Kafka: Producer dan Consumer.