kafka perf tweaks

Author: haileyok

cmd/tap/outbox.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"log/slog"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -231,7 +232,14 @@ func (o *Outbox) kafkaProduceAsync(evt *OutboxEvt) {
 		o.AckEvent(evt.ID)
 	}
 
-	if err := o.kafkaProducer.ProduceAsync(context.Background(), evt.Did, evt.Event, cb); err != nil {
+	var key string
+	// TODO: ordering bool
+	if true {
+		key = strconv.FormatUint(uint64(evt.ID), 10)
+	} else {
+		key = evt.Did
+	}
+	if err := o.kafkaProducer.ProduceAsync(context.Background(), key, evt.Event, cb); err != nil {
 		logger.Error("error queueing event for production", "error", err)
 		kafkaEventsProduced.WithLabelValues("error_queueing").Inc()
 	}

cmd/tap/tap.go

@@ -132,8 +132,8 @@ func NewTap(config TapConfig) (*Tap, error) {
 			logger.With("component", "kafkaProducer"),
 			config.KafkaBootstrapServers,
 			config.KafkaOutputTopic,
-			atkafka.WithEnsureTopic(true), // ensures that the topic has been created
-			atkafka.WithTopicPartitions(24),
+			atkafka.WithEnsureTopic(true),    // ensures that the topic has been created
+			atkafka.WithTopicPartitions(100), // TODO: partition count env var
 		)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create kafka producer: %w", err)