Repository: looplab/eventhorizon
Pull request #403 (2023-03-29 04:14:49)

Add Kafka event bus options for topic auto-create and partition count
### Description

Adds two configuration options for the Kafka event bus: `WithAutoCreateTopic(bool)` and `WithTopicPartitions(int)`.

### Affected Components

- Event Bus (Kafka)

### Related Issues

https://github.com/looplab/eventhorizon/issues/404

### Solution and Design

`WithAutoCreateTopic(bool)`: Specifies whether the Kafka topic for the event bus should be automatically created if it does not exist. Default remains `true`. If `true`, logic remains the same as before. If `false`, rather than calling `client.CreateTopics()`, a call to `client.Metadata()` is used to retrieve and verify the topic.

`WithTopicPartitions(int)`: Specifies the number of partitions to be created when creating the topic for the event bus. Default remains `5`.

### Steps to test and verify

1. Added integration tests for both options.



Patch:
diff --git a/eventbus/kafka/eventbus.go b/eventbus/kafka/eventbus.go
index 06ae239..43d7646 100644
--- a/eventbus/kafka/eventbus.go
+++ b/eventbus/kafka/eventbus.go
@@ -33,37 +33,41 @@ import (
 // to all matching registered handlers, in order of registration.
 type EventBus struct {
 	// TODO: Support multiple brokers.
-	addresses    []string
-	appID        string
-	topic        string
-	startOffset  int64
-	client       *kafka.Client
-	writer       *kafka.Writer
-	registered   map[eh.EventHandlerType]struct{}
-	registeredMu sync.RWMutex
-	errCh        chan error
-	cctx         context.Context
-	cancel       context.CancelFunc
-	wg           sync.WaitGroup
-	codec        eh.EventCodec
+	addresses       []string
+	appID           string
+	autoCreateTopic bool
+	topic           string
+	topicPartitions int
+	startOffset     int64
+	client          *kafka.Client
+	writer          *kafka.Writer
+	registered      map[eh.EventHandlerType]struct{}
+	registeredMu    sync.RWMutex
+	errCh           chan error
+	cctx            context.Context
+	cancel          context.CancelFunc
+	wg              sync.WaitGroup
+	codec           eh.EventCodec
 }
 
-// NewEventBus creates an EventBus, with optional GCP connection settings.
+// NewEventBus creates an EventBus, with optional settings.
 func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	addrSplit := strings.Split(addressList, ",")
 
 	b := &EventBus{
-		addresses:   addrSplit,
-		appID:       appID,
-		topic:       appID + "_events",
-		startOffset: kafka.LastOffset, // Default: Don't read old messages.
-		registered:  map[eh.EventHandlerType]struct{}{},
-		errCh:       make(chan error, 100),
-		cctx:        ctx,
-		cancel:      cancel,
-		codec:       &json.EventCodec{},
+		addresses:       addrSplit,
+		appID:           appID,
+		autoCreateTopic: true,
+		topic:           appID + "_events",
+		topicPartitions: 5,
+		startOffset:     kafka.LastOffset, // Default: Don't read old messages.
+		registered:      map[eh.EventHandlerType]struct{}{},
+		errCh:           make(chan error, 100),
+		cctx:            ctx,
+		cancel:          cancel,
+		codec:           &json.EventCodec{},
 	}
 
 	// Apply configuration options.
@@ -77,59 +81,112 @@ func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error
 		}
 	}
 
-	// Get or create the topic.
 	b.client = &kafka.Client{
 		Addr: kafka.TCP(addrSplit...),
 	}
 
-	var resp *kafka.CreateTopicsResponse
+	// Get or create the topic.
+	if b.autoCreateTopic {
+		if err := b.createTopic(); err != nil {
+			return nil, fmt.Errorf("error creating topic: %w", err)
+		}
+	} else if err := b.verifyTopic(); err != nil {
+		return nil, fmt.Errorf("error verifying topic: %w", err)
+	}
+
+	b.writer = &kafka.Writer{
+		Addr:         kafka.TCP(addrSplit...),
+		Topic:        b.topic,
+		BatchSize:    1,                // Write every event to the bus without delay.
+		RequiredAcks: kafka.RequireOne, // Stronger consistency.
+		Balancer:     &kafka.Hash{},    // Hash by aggregate ID.
+	}
+
+	return b, nil
+}
 
+// Creates the Kafka topic, with retries.
+func (b *EventBus) createTopic() error {
+	var resp *kafka.CreateTopicsResponse
 	var err error
 
 	for i := 0; i < 10; i++ {
 		resp, err = b.client.CreateTopics(context.Background(), &kafka.CreateTopicsRequest{
 			Topics: []kafka.TopicConfig{{
 				Topic:             b.topic,
-				NumPartitions:     5,
+				NumPartitions:     b.topicPartitions,
 				ReplicationFactor: 1,
 			}},
 		})
 
 		if errors.Is(err, kafka.BrokerNotAvailable) {
 			time.Sleep(5 * time.Second)
-
 			continue
 		} else if err != nil {
-			return nil, fmt.Errorf("error creating Kafka topic: %w", err)
+			return fmt.Errorf("error creating Kafka topic: %w", err)
 		} else {
 			break
 		}
 	}
 
 	if resp == nil {
-		return nil, fmt.Errorf("could not get/create Kafka topic in time: %w", err)
+		return fmt.Errorf("could not create Kafka topic in time: %w", err)
 	}
 
 	if topicErr, ok := resp.Errors[b.topic]; ok && topicErr != nil {
 		if !errors.Is(topicErr, kafka.TopicAlreadyExists) {
-			return nil, fmt.Errorf("invalid Kafka topic: %w", topicErr)
+			return fmt.Errorf("invalid Kafka topic: %w", topicErr)
 		}
 	}
 
-	b.writer = &kafka.Writer{
-		Addr:         kafka.TCP(addrSplit...),
-		Topic:        b.topic,
-		BatchSize:    1,                // Write every event to the bus without delay.
-		RequiredAcks: kafka.RequireOne, // Stronger consistency.
-		Balancer:     &kafka.Hash{},    // Hash by aggregate ID.
+	return nil
+}
+
+// Verifies that the Kafka topic exists, with retries.
+func (b *EventBus) verifyTopic() error {
+	var resp *kafka.MetadataResponse
+	var err error
+
+	for i := 0; i < 10; i++ {
+		resp, err = b.client.Metadata(context.Background(), &kafka.MetadataRequest{
+			Topics: []string{b.topic},
+		})
+
+		if errors.Is(err, kafka.BrokerNotAvailable) {
+			time.Sleep(5 * time.Second)
+			continue
+		} else if err != nil {
+			return fmt.Errorf("error getting Kafka topic: %w", err)
+		} else {
+			break
+		}
 	}
 
-	return b, nil
+	if resp == nil {
+		return fmt.Errorf("could not get Kafka topic in time: %w", err)
+	}
+
+	if len(resp.Topics) != 1 {
+		return fmt.Errorf("could not get Kafka topic: %w", err)
+	}
+
+	return nil
 }
 
 // Option is an option setter used to configure creation.
 type Option func(*EventBus) error
 
+// WithAutoCreateTopic specifies whether the event bus should
+// automatically create the topic if it does not exist.
+//
+// Defaults to: true
+func WithAutoCreateTopic(autoCreate bool) Option {
+	return func(b *EventBus) error {
+		b.autoCreateTopic = autoCreate
+		return nil
+	}
+}
+
 // WithCodec uses the specified codec for encoding events.
 func WithCodec(codec eh.EventCodec) Option {
 	return func(b *EventBus) error {
@@ -142,9 +199,10 @@ func WithCodec(codec eh.EventCodec) Option {
 // WithStartOffset sets the consumer group's offset to start at
 // Defaults to: LastOffset
 // Per the kafka client documentation
-//     StartOffset determines from whence the consumer group should begin
-//     consuming when it finds a partition without a committed offset.  If
-//     non-zero, it must be set to one of FirstOffset or LastOffset.
+//
+//	StartOffset determines from whence the consumer group should begin
+//	consuming when it finds a partition without a committed offset.  If
+//	non-zero, it must be set to one of FirstOffset or LastOffset.
 func WithStartOffset(startOffset int64) Option {
 	return func(b *EventBus) error {
 		b.startOffset = startOffset
@@ -152,7 +210,7 @@ func WithStartOffset(startOffset int64) Option {
 	}
 }
 
-// WithTopic uses the specified topic for the event bus topic name
+// WithTopic uses the specified topic for the event bus topic name.
 //
 // Defaults to: appID + "_events"
 func WithTopic(topic string) Option {
@@ -162,6 +220,24 @@ func WithTopic(topic string) Option {
 	}
 }
 
+// WithTopicPartitions uses the specified number of
+// partitions when creating the event bus topic.
+//
+// This option is ignored if the topic already exists
+// or if automatic topic creation is disabled.
+//
+// Defaults to: 5
+func WithTopicPartitions(numPartitions int) Option {
+	return func(b *EventBus) error {
+		if numPartitions < 1 {
+			return errors.New("number of partitions must be greater than 0")
+		}
+
+		b.topicPartitions = numPartitions
+		return nil
+	}
+}
+
 // HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.
 func (b *EventBus) HandlerType() eh.EventHandlerType {
 	return "eventbus"


Test patch:
diff --git a/eventbus/kafka/eventbus_test.go b/eventbus/kafka/eventbus_test.go
index 8d21a88..451f63b 100644
--- a/eventbus/kafka/eventbus_test.go
+++ b/eventbus/kafka/eventbus_test.go
@@ -17,6 +17,7 @@ package kafka
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"os"
 	"testing"
@@ -126,6 +127,33 @@ func newTestEventBus(appID string, options ...Option) (eh.EventBus, string, erro
 	return bus, appID, nil
 }
 
+func TestWithAutoCreateTopic(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	testCases := map[string]struct {
+		autoCreate bool
+	}{
+		"true":  {true},
+		"false": {false},
+	}
+
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			eb, _, err := newTestEventBus("", WithAutoCreateTopic(tc.autoCreate))
+			if err != nil {
+				t.Fatalf("expected no error, got: %s", err.Error())
+			}
+
+			underlyingEb := eb.(*EventBus)
+			if want, got := tc.autoCreate, underlyingEb.autoCreateTopic; want != got {
+				t.Fatalf("expected autoCreateTopic to be %t, got: %t", want, got)
+			}
+		})
+	}
+}
+
 func TestWithStartOffset(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
@@ -179,3 +207,36 @@ func TestWithTopic(t *testing.T) {
 		})
 	}
 }
+
+func TestWithTopicPatitions(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	invalidErr := errors.New("number of partitions must be greater than 0")
+
+	testCases := map[string]struct {
+		numPartitions int
+		expectError   error
+	}{
+		"valid number":           {42, nil},
+		"exactly 2":              {2, nil},
+		"exactly 1":              {1, nil},
+		"exactly 0":              {0, invalidErr},
+		"another invalid number": {-42, invalidErr},
+	}
+
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			eb, _, err := newTestEventBus("", WithTopicPartitions(tc.numPartitions))
+			if want, got := tc.expectError, err; want != got {
+				t.Fatalf("expected errors to be equal, want %s, got: %s", want, got)
+			}
+
+			underlyingEb := eb.(*EventBus)
+			if want, got := tc.numPartitions, underlyingEb.topicPartitions; want != got {
+				t.Fatalf("expected number of topic partitions to be equal, want %d, got: %d", want, got)
+			}
+		})
+	}
+}