Switch to kafka :^)

14 files changed, 131 insertions, 96 deletions

diff --git a/MODULE.bazel b/MODULE.bazel
index c9c22d2..dfc1f31 100644
--- a/MODULE.bazel
+++ b/MODULE.bazel
@@ -21,13 +21,12 @@ oci.pull(
         "linux/arm64",
     ],
 )
-
 use_repo(oci, "distroless_base")
 
 go_deps = use_extension("@bazel_gazelle//:extensions.bzl", "go_deps")
 go_deps.from_file(go_mod = "//:go.mod")
-
 use_repo(
     go_deps,
     "com_github_rabbitmq_amqp091_go",
+    "com_github_segmentio_kafka_go",
 )
diff --git a/README.md b/README.md
index 3026437..e370961 100644
--- a/README.md
+++ b/README.md
@@ -4,6 +4,6 @@ helm --namespace metallb-system install --create-namespace metallb metallb/metal
 kubectl apply -f metallb_config.yml
 kubectl apply -f metallb_announce.yml
 
-helm install rabbitmq oci://registry-1.docker.io/bitnamicharts/rabbitmq
+helm install kafka oci://registry-1.docker.io/bitnamicharts/kafka --set controller.replicaCount=1,controller.livenessProbe.initialDelaySeconds=120
 
 to deploy, execute the ./deploy.sh script
diff --git a/go.mod b/go.mod
index bac0b7f..4c09677 100644
--- a/go.mod
+++ b/go.mod
@@ -3,3 +3,9 @@ module github.com/Baitinq/fs-tracer-backend
 go 1.22.2
 
 require github.com/rabbitmq/amqp091-go v1.9.0
+
+require (
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+	github.com/segmentio/kafka-go v0.4.47
+)
diff --git a/go.sum b/go.sum
index 00c627d..487ba89 100644
--- a/go.sum
+++ b/go.sum
@@ -1,16 +1,65 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rabbitmq/amqp091-go v1.9.0 h1:qrQtyzB4H8BQgEuJwhmVQqVHB9O4+MNDJCCAcpc3Aoo=
 github.com/rabbitmq/amqp091-go v1.9.0/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc=
+github.com/segmentio/kafka-go v0.4.47 h1:IqziR4pA3vrZq7YdRxaT3w1/5fvIH5qpCwstUanQQB0=
+github.com/segmentio/kafka-go v0.4.47/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/k8s/payload-processor/templates/deployment.yaml b/k8s/payload-processor/templates/deployment.yaml
index c86398d..b9ab76e 100644
--- a/k8s/payload-processor/templates/deployment.yaml
+++ b/k8s/payload-processor/templates/deployment.yaml
@@ -34,11 +34,11 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           env:
-          - name: RABBITMQ_PASSWORD
+          - name: KAFKA_PASSWORD
             valueFrom:
               secretKeyRef:
-                name: rabbitmq
-                key: rabbitmq-password
+                name: kafka-user-passwords
+                key: client-passwords
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}
diff --git a/k8s/rest-api/templates/deployment.yaml b/k8s/rest-api/templates/deployment.yaml
index 4db09db..5951f59 100644
--- a/k8s/rest-api/templates/deployment.yaml
+++ b/k8s/rest-api/templates/deployment.yaml
@@ -48,11 +48,11 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           env:
-          - name: RABBITMQ_PASSWORD
+          - name: KAFKA_PASSWORD
             valueFrom:
               secretKeyRef:
-                name: rabbitmq
-                key: rabbitmq-password
+                name: kafka-user-passwords
+                key: client-passwords
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}
diff --git a/src/payload-processor/cmd/BUILD.bazel b/src/payload-processor/cmd/BUILD.bazel
index 515c17e..a11c89c 100644
--- a/src/payload-processor/cmd/BUILD.bazel
+++ b/src/payload-processor/cmd/BUILD.bazel
@@ -9,7 +9,8 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//src/payload-processor/processor",
-        "@com_github_rabbitmq_amqp091_go//:amqp091-go",
+        "@com_github_segmentio_kafka_go//:kafka-go",
+        "@com_github_segmentio_kafka_go//sasl/plain",
     ],
 )
 
diff --git a/src/payload-processor/cmd/main.go b/src/payload-processor/cmd/main.go
index 1c9bfcb..8e158e9 100644
--- a/src/payload-processor/cmd/main.go
+++ b/src/payload-processor/cmd/main.go
@@ -1,43 +1,34 @@
 package main
 
 import (
-	"fmt"
 	"log"
 	"os"
+	"time"
 
 	"github.com/Baitinq/fs-tracer-backend/src/payload-processor/processor"
-	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl/plain"
 )
 
 func main() {
-	rabbitmq_password, ok := os.LookupEnv("RABBITMQ_PASSWORD")
+	kafka_password, ok := os.LookupEnv("KAFKA_PASSWORD")
 	if !ok {
-		log.Fatal("RABBITMQ_PASSWORD not set")
+		log.Fatal("KAFKA_PASSWORD not set")
 	}
-	log.Println("RabbitMQ password", rabbitmq_password)
-	conn, err := amqp.Dial(fmt.Sprintf("amqp://user:%s@rabbitmq:5672/", rabbitmq_password))
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer conn.Close()
-
-	ch, err := conn.Channel()
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer ch.Close()
-
-	q, err := ch.QueueDeclare(
-		"hello", // name
-		false,   // durable
-		false,   // delete when unused
-		false,   // exclusive
-		false,   // no-wait
-		nil,     // arguments
-	)
-	if err != nil {
-		log.Fatal(err)
-	}
-	processor := processor.NewProcessor(ch, q.Name)
+	kafka_reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers: []string{"kafka.default.svc.cluster.local:9092"},
+		Dialer: &kafka.Dialer{
+			SASLMechanism: plain.Mechanism{
+				Username: "user1",
+				Password: kafka_password,
+			},
+			Timeout:   10 * time.Second,
+			DualStack: true,
+		},
+		Topic:     "topic-A",
+		Partition: 0,    //TODO: What
+		MaxBytes:  10e6, // 10MB
+	})
+	processor := processor.NewProcessor(kafka_reader)
 	processor.ProcessMessages()
 }
diff --git a/src/payload-processor/processor/BUILD.bazel b/src/payload-processor/processor/BUILD.bazel
index 64e3ac2..123059d 100644
--- a/src/payload-processor/processor/BUILD.bazel
+++ b/src/payload-processor/processor/BUILD.bazel
@@ -5,5 +5,5 @@ go_library(
     srcs = ["processor.go"],
     importpath = "github.com/Baitinq/fs-tracer-backend/src/payload-processor/processor",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_rabbitmq_amqp091_go//:amqp091-go"],
+    deps = ["@com_github_segmentio_kafka_go//:kafka-go"],
 )
diff --git a/src/payload-processor/processor/processor.go b/src/payload-processor/processor/processor.go
index 3eb089b..435a954 100644
--- a/src/payload-processor/processor/processor.go
+++ b/src/payload-processor/processor/processor.go
@@ -1,39 +1,34 @@
 package processor
 
 import (
+	"context"
+	"fmt"
 	"log"
 
-	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/segmentio/kafka-go"
 )
 
 type Processor struct {
-	ch        *amqp.Channel
-	queueName string
+	kafka_reader *kafka.Reader
 }
 
-func NewProcessor(ch *amqp.Channel, queueName string) Processor {
+func NewProcessor(kafka_reader *kafka.Reader) Processor {
 	log.Println("Created processor")
 	return Processor{
-		ch:        ch,
-		queueName: queueName,
+		kafka_reader: kafka_reader,
 	}
 }
 
 func (p Processor) ProcessMessages() {
-	msgs, err := p.ch.Consume(
-		p.queueName, // queue
-		"",          // consumer
-		true,        // auto-ack
-		false,       // exclusive
-		false,       // no-local
-		false,       // no-wait
-		nil,         // args
-	)
-	if err != nil {
-		log.Fatal("Failed to register a consumer:", err)
+	for {
+		m, err := p.kafka_reader.ReadMessage(context.Background())
+		if err != nil {
+			break
+		}
+		fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
 	}
 
-	for msg := range msgs {
-		log.Printf("Received a message: %s", msg.Body)
+	if err := p.kafka_reader.Close(); err != nil {
+		log.Fatal("failed to close reader:", err)
 	}
 }
diff --git a/src/rest-api/cmd/BUILD.bazel b/src/rest-api/cmd/BUILD.bazel
index 842f6f9..2a142c7 100644
--- a/src/rest-api/cmd/BUILD.bazel
+++ b/src/rest-api/cmd/BUILD.bazel
@@ -9,7 +9,8 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//src/rest-api/handler",
-        "@com_github_rabbitmq_amqp091_go//:amqp091-go",
+        "@com_github_segmentio_kafka_go//:kafka-go",
+        "@com_github_segmentio_kafka_go//sasl/plain",
     ],
 )
 
diff --git a/src/rest-api/cmd/main.go b/src/rest-api/cmd/main.go
index f6d34e0..469957f 100644
--- a/src/rest-api/cmd/main.go
+++ b/src/rest-api/cmd/main.go
@@ -7,39 +7,31 @@ import (
 	"os"
 
 	"github.com/Baitinq/fs-tracer-backend/src/rest-api/handler"
-	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl/plain"
 )
 
 func main() {
-	rabbitmq_password, ok := os.LookupEnv("RABBITMQ_PASSWORD")
+	kafka_password, ok := os.LookupEnv("KAFKA_PASSWORD")
 	if !ok {
-		log.Fatal("RABBITMQ_PASSWORD not set")
+		log.Fatal("KAFKA_PASSWORD not set")
 	}
-	log.Println("RabbitMQ password", rabbitmq_password)
-	conn, err := amqp.Dial(fmt.Sprintf("amqp://user:%s@rabbitmq:5672/", rabbitmq_password))
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer conn.Close()
 
-	ch, err := conn.Channel()
-	if err != nil {
-		log.Fatal(err)
+	kafka_writer := &kafka.Writer{
+		Addr: kafka.TCP("kafka.default.svc.cluster.local:9092"),
+		Transport: &kafka.Transport{
+			SASL: plain.Mechanism{
+				Username: "user1",
+				Password: kafka_password,
+			},
+		},
+		Topic:    "topic-A",
+		Balancer: &kafka.LeastBytes{},
+		// Async:                  true, //TODO: Creat the topic beforehand, if not this doesnt work
+		AllowAutoTopicCreation: true,
 	}
-	defer ch.Close()
 
-	q, err := ch.QueueDeclare(
-		"hello", // name
-		false,   // durable
-		false,   // delete when unused
-		false,   // exclusive
-		false,   // no-wait
-		nil,     // arguments
-	)
-	if err != nil {
-		log.Fatal(err)
-	}
-	handler := handler.NewHandler(ch, q.Name)
+	handler := handler.NewHandler(kafka_writer)
 
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
diff --git a/src/rest-api/handler/BUILD.bazel b/src/rest-api/handler/BUILD.bazel
index a5638f4..89adc69 100644
--- a/src/rest-api/handler/BUILD.bazel
+++ b/src/rest-api/handler/BUILD.bazel
@@ -5,5 +5,5 @@ go_library(
     srcs = ["handler.go"],
     importpath = "github.com/Baitinq/fs-tracer-backend/src/rest-api/handler",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_rabbitmq_amqp091_go//:amqp091-go"],
+    deps = ["@com_github_segmentio_kafka_go//:kafka-go"],
 )
diff --git a/src/rest-api/handler/handler.go b/src/rest-api/handler/handler.go
index b9bb80e..8985b43 100644
--- a/src/rest-api/handler/handler.go
+++ b/src/rest-api/handler/handler.go
@@ -8,18 +8,16 @@ import (
 	"net/http"
 	"time"
 
-	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/segmentio/kafka-go"
 )
 
 type Handler struct {
-	ch        *amqp.Channel
-	queueName string
+	kafka_writer *kafka.Writer
 }
 
-func NewHandler(ch *amqp.Channel, queueName string) Handler {
+func NewHandler(kafka_writer *kafka.Writer) Handler {
 	return Handler{
-		ch:        ch,
-		queueName: queueName,
+		kafka_writer: kafka_writer,
 	}
 }
 
@@ -34,11 +32,14 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 
-	go h.ch.PublishWithContext(ctx, "", h.queueName, false, false, amqp.Publishing{
-		ContentType: "text/plain",
-		Body:        []byte(body),
+	err = h.kafka_writer.WriteMessages(ctx, kafka.Message{
+		Key:   []byte("key-A"),
+		Value: []byte(body),
 	})
+	if err != nil {
+		log.Fatal(err)
+	}
 
 	fmt.Fprint(w, "Hello, World!", string(bytes))
-	log.Println("Request received", r.RemoteAddr, string(bytes))
+	log.Println("Request received :)", r.RemoteAddr, string(bytes))
 }