consum correctly

Author: rajsinghtech

examples/kustomization/kafka.yaml

@@ -18,10 +18,14 @@ spec:
         port: 9092
         type: internal
         tls: false
+        configuration:
+          useServiceDnsDomain: true
       - name: tls
         port: 9093
         type: internal
         tls: true
+        configuration:
+          useServiceDnsDomain: true
     config:
       offsets.topic.replication.factor: 3
       transaction.state.log.replication.factor: 3

telemetry/consumer.go

@@ -1,7 +1,6 @@
 package telemetry
 
 import (
-	"encoding/base64"
 	"fmt"
 	"log"
 	"os"
@@ -84,15 +83,10 @@ func (c *Consumer) Start() error {
 }
 
 func (c *Consumer) handleMessage(msg *kafka.Message) error {
-	// The message value should be a base64 encoded protobuf
-	data, err := base64.StdEncoding.DecodeString(string(msg.Value))
-	if err != nil {
-		return fmt.Errorf("failed to decode base64 message: %v", err)
-	}
-
-	// Unmarshal the protobuf message
+	// The message value should be raw protobuf data
+	// Unmarshal the protobuf message directly
 	payload := &protos.Payload{}
-	if err := proto.Unmarshal(data, payload); err != nil {
+	if err := proto.Unmarshal(msg.Value, payload); err != nil {
 		return fmt.Errorf("failed to unmarshal protobuf: %v", err)
 	}