Repository: looplab/eventhorizon
Pull request #334 (2021-09-06 08:38:28)

Event bus performance testing
### Description

Event bus performance testing.

### Affected Components

- Event Bus tests

### Related Issues

### Solution and Design

### Steps to test and verify


coveralls: 
[![Coverage Status](https://coveralls.io/builds/42671438/badge)](https://coveralls.io/builds/42671438)

Coverage increased (+0.1%) to 69.588% when pulling **7b18e6f08a612b2246f60b328e92ff663a98a3ac on maxekman:fix/eventbus-performance-testing** into **e82556fcd0b5c8697599bd099801e061082cf7cb on looplab:main**.


Patch:
diff --git a/docker-compose.yml b/docker-compose.yml
index 7b77a4c..d83b277 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -15,14 +15,12 @@ services:
       - redis
       - jetstream
     environment:
-      MONGODB_ADDR: "mongodb:27017"
-      PUBSUB_EMULATOR_HOST: "gpubsub:8793"
-      KAFKA_ADDR: "kafka:9092"
-      REDIS_ADDR: "redis:6379"
-      NATS_ADDR: "jetstream:4222"
-    command:
-      - -c
-      - make test test_integration
+      MONGODB_ADDR: mongodb:27017
+      PUBSUB_EMULATOR_HOST: gpubsub:8793
+      KAFKA_ADDR: kafka:9092
+      REDIS_ADDR: redis:6379
+      NATS_ADDR: jetstream:4222
+    command: [-c, make test test_integration]
 
   # MongoDB replica set setup.
   mongodb:
@@ -32,7 +30,7 @@ services:
       - mongodb-rs
     networks:
       default:
-    entrypoint: ["bash", "/scripts/mongodb-rs-setup.sh"]
+    entrypoint: [bash, /scripts/mongodb-rs-setup.sh]
     volumes:
       - ./scripts/mongodb-rs-setup.sh:/scripts/mongodb-rs-setup.sh
 
@@ -40,8 +38,8 @@ services:
     image: mongo:4.4
     restart: always
     ports:
-      - "27017:27017"
-    command: ["--replSet", "rs0", "--bind_ip_all"]
+      - 27017:27017
+    command: [--replSet, rs0, --bind_ip_all]
     healthcheck:
       test: test $$(echo "rs.status().ok" | mongo --quiet) -eq 1
       interval: 10s
@@ -50,21 +48,16 @@ services:
       - mongodb:/data/db
 
   gpubsub:
-    image: gcr.io/google.com/cloudsdktool/cloud-sdk:326.0.0-emulators
+    image: gcr.io/google.com/cloudsdktool/cloud-sdk:355.0.0-emulators
     ports:
-      - "8793:8793"
+      - 8793:8793
     entrypoint:
-      - gcloud
-      - beta
-      - emulators
-      - pubsub
-      - start
-      - "--host-port=0.0.0.0:8793"
+      [gcloud, beta, emulators, pubsub, start, --host-port=0.0.0.0:8793]
 
   zookeeper:
     image: bitnami/zookeeper:3.6.2
     ports:
-      - "2181:2181"
+      - 2181:2181
     environment:
       ALLOW_ANONYMOUS_LOGIN: "yes"
 
@@ -73,31 +66,28 @@ services:
     depends_on:
       - zookeeper
     ports:
-      - "9093:9093"
+      - 9093:9093
     environment:
-      KAFKA_BROKER_ID: "1"
-      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
       ALLOW_PLAINTEXT_LISTENER: "yes"
       KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:9093
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:9093
-      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
-      KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: "5"
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 5
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
 
   redis:
     image: redis:6.2-alpine3.13
     ports:
-      - "6379:6379"
-    entrypoint:
-      - redis-server
-      - "--appendonly yes"
+      - 6379:6379
+    entrypoint: [redis-server, --appendonly yes]
 
   jetstream:
     image: synadia/jsm:nightly
     ports:
-      - "4222:4222"
-      - "8222:8222"
-    command:
-      - server
+      - 4222:4222
+      - 8222:8222
+    command: [server]
diff --git a/eventbus/local/eventbus.go b/eventbus/local/eventbus.go
index ee667fd..1b21cc6 100644
--- a/eventbus/local/eventbus.go
+++ b/eventbus/local/eventbus.go
@@ -26,7 +26,7 @@ import (
 )
 
 // DefaultQueueSize is the default queue size per handler for publishing events.
-var DefaultQueueSize = 10
+var DefaultQueueSize = 1000
 
 // EventBus is a local event bus that delegates handling of published events
 // to all matching registered handlers, in order of registration.
@@ -145,7 +145,7 @@ func (b *EventBus) handle(ctx context.Context, m eh.EventMatcher, h eh.EventHand
 		select {
 		case data := <-ch:
 			// Artificial delay to simulate network.
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(time.Millisecond)
 
 			event, ctx, err := b.codec.UnmarshalEvent(ctx, data)
 			if err != nil {


Test patch:
diff --git a/eventbus/acceptance_testing.go b/eventbus/acceptance_testing.go
index 77b5173..e2309fe 100644
--- a/eventbus/acceptance_testing.go
+++ b/eventbus/acceptance_testing.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/rand"
 	"sync"
 	"testing"
 	"time"
@@ -30,18 +31,8 @@ import (
 	"github.com/looplab/eventhorizon/uuid"
 )
 
-// AcceptanceTest is the acceptance test that all implementations of EventBus
-// should pass. It should manually be called from a test case in each
-// implementation:
-//
-//   func TestEventBus(t *testing.T) {
-//       bus1 := NewEventBus()
-//       bus2 := NewEventBus()
-//       eventbus.AcceptanceTest(t, bus1, bus2)
-//   }
-//
-func AcceptanceTest(t *testing.T, bus1, bus2 eh.EventBus, timeout time.Duration) {
-	ctx, cancel := context.WithCancel(context.Background())
+func TestAddHandler(t *testing.T, bus1 eh.EventBus) {
+	ctx := context.Background()
 
 	// Error on nil matcher.
 	if err := bus1.AddHandler(ctx, nil, mocks.NewEventHandler("no-matcher")); err != eh.ErrMissingMatcher {
@@ -60,7 +51,20 @@ func AcceptanceTest(t *testing.T, bus1, bus2 eh.EventBus, timeout time.Duration)
 	if err := bus1.AddHandler(ctx, eh.MatchAll{}, mocks.NewEventHandler("multi")); err != eh.ErrHandlerAlreadyAdded {
 		t.Error("the error should be correct:", err)
 	}
+}
 
+// AcceptanceTest is the acceptance test that all implementations of EventBus
+// should pass. It should manually be called from a test case in each
+// implementation:
+//
+//   func TestEventBus(t *testing.T) {
+//       bus1 := NewEventBus()
+//       bus2 := NewEventBus()
+//       eventbus.AcceptanceTest(t, bus1, bus2)
+//   }
+//
+func AcceptanceTest(t *testing.T, bus1, bus2 eh.EventBus, timeout time.Duration) {
+	ctx, cancel := context.WithCancel(context.Background())
 	ctx = mocks.WithContextOne(ctx, "testval")
 
 	// Without handler.
@@ -263,10 +267,34 @@ func checkBusErrors(t *testing.T, bus eh.EventBus) {
 
 // LoadTest is a load test for an event bus implementation.
 func LoadTest(t *testing.T, bus eh.EventBus) {
+	benchmark(&testAsBench{t}, bus, 20, 10, 1000)
+}
+
+// Benchmark is a benchmark for an event bus implementation.
+func Benchmark(b *testing.B, bus eh.EventBus) {
+	benchmark(b, bus, 20, 10, b.N)
+}
+
+type bench interface {
+	Log(args ...interface{})
+	Error(args ...interface{})
+	Fatal(args ...interface{})
+
+	ResetTimer()
+	StopTimer()
+}
+
+type testAsBench struct {
+	*testing.T
+}
+
+func (b *testAsBench) ResetTimer() {}
+func (b *testAsBench) StopTimer()  {}
+
+func benchmark(t bench, bus eh.EventBus, numAggregates, numHandlers, numEvents int) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	// Add handlers.
-	handlers := make([]*mocks.EventHandler, 100)
+	handlers := make([]*mocks.EventHandler, numHandlers)
 	for i := range handlers {
 		h := mocks.NewEventHandler(fmt.Sprintf("handler-%d", i))
 		if err := bus.AddHandler(ctx, eh.MatchAll{}, h); err != nil {
@@ -275,85 +303,60 @@ func LoadTest(t *testing.T, bus eh.EventBus) {
 		handlers[i] = h
 	}
 
-	time.Sleep(3 * time.Second) // Need to wait here for handlers to be added.
+	time.Sleep(time.Second) // Need to wait here for handler to be added.
 	t.Log("setup complete")
+	t.Log("num events:", numEvents)
 
-	// Start receiving.
 	var wg sync.WaitGroup
 	for _, h := range handlers {
 		wg.Add(1)
 		go func(h *mocks.EventHandler) {
-			select {
-			case <-h.Recv:
-			case <-time.After(30 * time.Second):
-				t.Error("did not receive message within timeout")
+			defer wg.Done()
+			numEventsReceived := 0
+			for {
+				if numEventsReceived == numEvents {
+					return
+				}
+				select {
+				case <-h.Recv:
+					numEventsReceived++
+					continue
+				case <-ctx.Done():
+					return
+				case <-time.After(30 * time.Second):
+					t.Error("did not receive message within timeout")
+					return
+				}
 			}
-			wg.Done()
 		}(h)
 	}
 
-	// Handle event.
-	id := uuid.New()
-	timestamp := time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)
-	event1 := eh.NewEvent(
-		mocks.EventType, &mocks.EventData{Content: "event1"}, timestamp,
-		eh.ForAggregate(mocks.AggregateType, id, 1))
-	if err := bus.HandleEvent(ctx, event1); err != nil {
-		t.Error("there should be no error:", err)
+	var aggregates []*struct {
+		id      uuid.UUID
+		version int
 	}
-
-	// Wait for all handlers to have received their event.
-	wg.Wait()
-
-	// Cancel all handlers and wait.
-	cancel()
-	bus.Wait()
-}
-
-// Benchmark is a benchmark for an event bus implementation.
-func Benchmark(b *testing.B, bus eh.EventBus) {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	h := mocks.NewEventHandler("handler")
-	if err := bus.AddHandler(ctx, eh.MatchAll{}, h); err != nil {
-		b.Fatal("there should be no error:", err)
+	for i := 0; i < numAggregates; i++ {
+		aggregates = append(aggregates, &struct {
+			id      uuid.UUID
+			version int
+		}{uuid.New(), 0})
 	}
 
-	time.Sleep(3 * time.Second) // Need to wait here for handler to be added.
-	b.Log("setup complete")
-	b.Log("num iterations:", b.N)
-	b.ResetTimer()
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		// Wait for all events in batch.
-		for n := 0; n < b.N; n++ {
-			select {
-			case <-h.Recv:
-				continue
-			case <-ctx.Done():
-				return
-			case <-time.After(30 * time.Second):
-				b.Error("did not receive message within timeout")
-				return
-			}
-		}
-	}()
-
-	id := uuid.New()
-	for n := 0; n < b.N; n++ {
+	t.ResetTimer()
+	for n := 0; n < numEvents; n++ {
+		a := aggregates[rand.Intn(len(aggregates))]
+		a.version++
+		timestamp := time.Date(2009, time.November, 10, 23, n, 0, 0, time.UTC)
 		e := eh.NewEvent(
-			mocks.EventType, &mocks.EventData{Content: fmt.Sprintf("event%d", n)}, time.Now(),
-			eh.ForAggregate(mocks.AggregateType, id, n+1))
+			mocks.EventType, &mocks.EventData{Content: fmt.Sprintf("event%d", n)}, timestamp,
+			eh.ForAggregate(mocks.AggregateType, a.id, a.version))
 		if err := bus.HandleEvent(ctx, e); err != nil {
-			b.Error("there should be no error:", err)
+			t.Error("there should be no error:", err)
 		}
 	}
 
 	wg.Wait() // Wait for all events to be received.
-	b.StopTimer()
+	t.StopTimer()
 
 	// Cancel handler and wait.
 	cancel()
diff --git a/eventbus/gcp/eventbus_test.go b/eventbus/gcp/eventbus_test.go
index a35d517..ff3395a 100644
--- a/eventbus/gcp/eventbus_test.go
+++ b/eventbus/gcp/eventbus_test.go
@@ -17,40 +17,45 @@ package gcp
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
+	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 )
 
-func TestEventBusIntegration(t *testing.T) {
+func TestAddHandlerIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	if os.Getenv("PUBSUB_EMULATOR_HOST") == "" {
-		os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8793")
+	bus1, _, err := newTestEventBus("")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
 	}
 
-	// Get a random app ID.
-	b := make([]byte, 8)
-	if _, err := rand.Read(b); err != nil {
-		t.Fatal(err)
+	eventbus.TestAddHandler(t, bus1)
+}
+
+func TestEventBusIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
 	}
-	appID := "app-" + hex.EncodeToString(b)
 
-	bus1, err := NewEventBus("project_id", appID)
+	bus1, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
-	bus2, err := NewEventBus("project_id", appID)
+	bus2, _, err := newTestEventBus(appID)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
 }
 
@@ -59,43 +64,46 @@ func TestEventBusLoadtest(t *testing.T) {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	if os.Getenv("PUBSUB_EMULATOR_HOST") == "" {
-		os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8793")
-	}
-
-	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		t.Fatal(err)
-	}
-	appID := "app-" + hex.EncodeToString(bts)
-
-	bus, err := NewEventBus("project_id", appID)
+	bus, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.LoadTest(t, bus)
 }
 
 func BenchmarkEventBus(b *testing.B) {
+	bus, appID, err := newTestEventBus("")
+	if err != nil {
+		b.Fatal("there should be no error:", err)
+	}
+
+	b.Logf("using topic: %s_events", appID)
+
+	eventbus.Benchmark(b, bus)
+}
+
+func newTestEventBus(appID string) (eh.EventBus, string, error) {
 	// Connect to localhost if not running inside docker
 	if os.Getenv("PUBSUB_EMULATOR_HOST") == "" {
 		os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8793")
 	}
 
 	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		b.Fatal(err)
+	if appID == "" {
+		bts := make([]byte, 8)
+		if _, err := rand.Read(bts); err != nil {
+			return nil, "", fmt.Errorf("could not randomize app ID: %w", err)
+		}
+		appID = "app-" + hex.EncodeToString(bts)
 	}
-	appID := "app-" + hex.EncodeToString(bts)
 
 	bus, err := NewEventBus("project_id", appID)
 	if err != nil {
-		b.Fatal("there should be no error:", err)
+		return nil, "", fmt.Errorf("could not create event bus: %w", err)
 	}
 
-	eventbus.Benchmark(b, bus)
+	return bus, appID, nil
 }
diff --git a/eventbus/jetstream/eventbus_test.go b/eventbus/jetstream/eventbus_test.go
index 2a31b33..63172c2 100644
--- a/eventbus/jetstream/eventbus_test.go
+++ b/eventbus/jetstream/eventbus_test.go
@@ -17,42 +17,45 @@ package jetstream
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
+	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 )
 
-func TestEventBusIntegration(t *testing.T) {
+func TestAddHandlerIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
 
-	// Enable testing with Docker, default to local testing.
-	addr := os.Getenv("NATS_ADDR")
-	if addr == "" {
-		addr = "localhost:4222"
+	bus1, _, err := newTestEventBus("")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
 	}
-	url := "nats://" + addr
 
-	// Get a random app ID.
-	b := make([]byte, 4)
-	if _, err := rand.Read(b); err != nil {
-		t.Fatal(err)
+	eventbus.TestAddHandler(t, bus1)
+}
+
+func TestEventBusIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
 	}
-	appID := "app-" + hex.EncodeToString(b)
 
-	bus1, err := NewEventBus(url, appID)
+	bus1, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
-	bus2, err := NewEventBus(url, appID)
+	bus2, _, err := newTestEventBus(appID)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
 }
 
@@ -61,29 +64,28 @@ func TestEventBusLoadtest(t *testing.T) {
 		t.Skip("skipping integration test")
 	}
 
-	// Enable testing with Docker, default to local testing.
-	addr := os.Getenv("NATS_ADDR")
-	if addr == "" {
-		addr = "localhost:4222"
-	}
-	url := "nats://" + addr
-
-	// Get a random app ID.
-	bts := make([]byte, 4)
-	if _, err := rand.Read(bts); err != nil {
-		t.Fatal(err)
-	}
-	appID := "app-" + hex.EncodeToString(bts)
-
-	bus, err := NewEventBus(url, appID)
+	bus, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.LoadTest(t, bus)
 }
 
 func BenchmarkEventBus(b *testing.B) {
+	bus, appID, err := newTestEventBus("")
+	if err != nil {
+		b.Fatal("there should be no error:", err)
+	}
+
+	b.Logf("using topic: %s_events", appID)
+
+	eventbus.Benchmark(b, bus)
+}
+
+func newTestEventBus(appID string) (eh.EventBus, string, error) {
 	// Enable testing with Docker, default to local testing.
 	addr := os.Getenv("NATS_ADDR")
 	if addr == "" {
@@ -92,16 +94,18 @@ func BenchmarkEventBus(b *testing.B) {
 	url := "nats://" + addr
 
 	// Get a random app ID.
-	bts := make([]byte, 4)
-	if _, err := rand.Read(bts); err != nil {
-		b.Fatal(err)
+	if appID == "" {
+		b := make([]byte, 4)
+		if _, err := rand.Read(b); err != nil {
+			return nil, "", fmt.Errorf("could not randomize app ID: %w", err)
+		}
+		appID = "app-" + hex.EncodeToString(b)
 	}
-	appID := "app-" + hex.EncodeToString(bts)
 
 	bus, err := NewEventBus(url, appID)
 	if err != nil {
-		b.Fatal("there should be no error:", err)
+		return nil, "", fmt.Errorf("could not create event bus: %w", err)
 	}
 
-	eventbus.Benchmark(b, bus)
+	return bus, appID, nil
 }
diff --git a/eventbus/kafka/eventbus_test.go b/eventbus/kafka/eventbus_test.go
index a582fcc..7c4e5df 100644
--- a/eventbus/kafka/eventbus_test.go
+++ b/eventbus/kafka/eventbus_test.go
@@ -17,43 +17,45 @@ package kafka
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
+	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 )
 
-func TestEventBusIntegration(t *testing.T) {
+func TestAddHandlerIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	addr := os.Getenv("KAFKA_ADDR")
-	if addr == "" {
-		addr = "localhost:9093"
+	bus1, _, err := newTestEventBus("")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
 	}
 
-	// Get a random app ID.
-	b := make([]byte, 8)
-	if _, err := rand.Read(b); err != nil {
-		t.Fatal(err)
-	}
-	appID := "app-" + hex.EncodeToString(b)
+	eventbus.TestAddHandler(t, bus1)
+}
 
-	t.Logf("using topic: %s_events", appID)
+func TestEventBusIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
 
-	bus1, err := NewEventBus(addr, appID)
+	bus1, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
-	bus2, err := NewEventBus(addr, appID)
+	bus2, _, err := newTestEventBus(appID)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
 }
 
@@ -62,28 +64,28 @@ func TestEventBusLoadtest(t *testing.T) {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	addr := os.Getenv("KAFKA_ADDR")
-	if addr == "" {
-		addr = "localhost:9093"
-	}
-
-	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		t.Fatal(err)
-	}
-	appID := "app-" + hex.EncodeToString(bts)
-
-	bus, err := NewEventBus(addr, appID)
+	bus, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using topic: %s_events", appID)
+
 	eventbus.LoadTest(t, bus)
 }
 
 func BenchmarkEventBus(b *testing.B) {
+	bus, appID, err := newTestEventBus("")
+	if err != nil {
+		b.Fatal("there should be no error:", err)
+	}
+
+	b.Logf("using topic: %s_events", appID)
+
+	eventbus.Benchmark(b, bus)
+}
+
+func newTestEventBus(appID string) (eh.EventBus, string, error) {
 	// Connect to localhost if not running inside docker
 	addr := os.Getenv("KAFKA_ADDR")
 	if addr == "" {
@@ -91,16 +93,18 @@ func BenchmarkEventBus(b *testing.B) {
 	}
 
 	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		b.Fatal(err)
+	if appID == "" {
+		b := make([]byte, 8)
+		if _, err := rand.Read(b); err != nil {
+			return nil, "", fmt.Errorf("could not randomize app ID: %w", err)
+		}
+		appID = "app-" + hex.EncodeToString(b)
 	}
-	appID := "app-" + hex.EncodeToString(bts)
 
 	bus, err := NewEventBus(addr, appID)
 	if err != nil {
-		b.Fatal("there should be no error:", err)
+		return nil, "", fmt.Errorf("could not create event bus: %w", err)
 	}
 
-	eventbus.Benchmark(b, bus)
+	return bus, appID, nil
 }
diff --git a/eventbus/local/eventbus_test.go b/eventbus/local/eventbus_test.go
index f1dae60..57efac8 100644
--- a/eventbus/local/eventbus_test.go
+++ b/eventbus/local/eventbus_test.go
@@ -21,6 +21,20 @@ import (
 	"github.com/looplab/eventhorizon/eventbus"
 )
 
+// NOTE: Not named "Integration" to enable running with the unit tests.
+func TestAddHandler(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	bus := NewEventBus()
+	if bus == nil {
+		t.Fatal("there should be a bus")
+	}
+
+	eventbus.TestAddHandler(t, bus)
+}
+
 // NOTE: Not named "Integration" to enable running with the unit tests.
 func TestEventBus(t *testing.T) {
 	group := NewGroup()
diff --git a/eventbus/redis/eventbus_test.go b/eventbus/redis/eventbus_test.go
index 454b7d0..cf13e15 100644
--- a/eventbus/redis/eventbus_test.go
+++ b/eventbus/redis/eventbus_test.go
@@ -17,42 +17,46 @@ package redis
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/go-redis/redis/v8"
+
+	eh "github.com/looplab/eventhorizon"
 	"github.com/looplab/eventhorizon/eventbus"
 )
 
-func TestEventBusIntegration(t *testing.T) {
+func TestAddHandlerIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	addr := os.Getenv("REDIS_ADDR")
-	if addr == "" {
-		addr = "localhost:6379"
+	bus1, _, err := newTestEventBus("")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
 	}
 
-	// Get a random app ID.
-	b := make([]byte, 8)
-	if _, err := rand.Read(b); err != nil {
-		t.Fatal(err)
+	eventbus.TestAddHandler(t, bus1)
+}
+func TestEventBusIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
 	}
-	appID := "app-" + hex.EncodeToString(b)
 
-	bus1, err := NewEventBus(addr, appID, "client1")
+	bus1, appID, err := newTestEventBus("")
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
-	bus2, err := NewEventBus(addr, appID, "client2")
+	bus2, _, err := newTestEventBus(appID)
 	if err != nil {
 		t.Fatal("there should be no error:", err)
 	}
 
+	t.Logf("using stream: %s_events", appID)
+
 	eventbus.AcceptanceTest(t, bus1, bus2, time.Second)
 }
 
@@ -61,33 +65,28 @@ func TestEventBusLoadtest(t *testing.T) {
 		t.Skip("skipping integration test")
 	}
 
-	// Connect to localhost if not running inside docker
-	addr := os.Getenv("REDIS_ADDR")
-	if addr == "" {
-		addr = "localhost:6379"
+	bus, appID, err := newTestEventBus("")
+	if err != nil {
+		t.Fatal("there should be no error:", err)
 	}
 
-	opts := &redis.Options{
-		Addr:     addr,
-		PoolSize: 200,
-	}
+	t.Logf("using stream: %s_events", appID)
 
-	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		t.Fatal(err)
-	}
-	appID := "app-" + hex.EncodeToString(bts)
+	eventbus.LoadTest(t, bus)
+}
 
-	bus, err := NewEventBus(addr, appID, "client", WithRedisOptions(opts))
+func BenchmarkEventBus(b *testing.B) {
+	bus, appID, err := newTestEventBus("")
 	if err != nil {
-		t.Fatal("there should be no error:", err)
+		b.Fatal("there should be no error:", err)
 	}
 
-	eventbus.LoadTest(t, bus)
+	b.Logf("using stream: %s_events", appID)
+
+	eventbus.Benchmark(b, bus)
 }
 
-func BenchmarkEventBus(b *testing.B) {
+func newTestEventBus(appID string) (eh.EventBus, string, error) {
 	// Connect to localhost if not running inside docker
 	addr := os.Getenv("REDIS_ADDR")
 	if addr == "" {
@@ -100,16 +99,18 @@ func BenchmarkEventBus(b *testing.B) {
 	}
 
 	// Get a random app ID.
-	bts := make([]byte, 8)
-	if _, err := rand.Read(bts); err != nil {
-		b.Fatal(err)
+	if appID == "" {
+		bts := make([]byte, 8)
+		if _, err := rand.Read(bts); err != nil {
+			return nil, "", fmt.Errorf("could not randomize app ID: %w", err)
+		}
+		appID = "app-" + hex.EncodeToString(bts)
 	}
-	appID := "app-" + hex.EncodeToString(bts)
 
 	bus, err := NewEventBus(addr, appID, "client", WithRedisOptions(opts))
 	if err != nil {
-		b.Fatal("there should be no error:", err)
+		return nil, "", fmt.Errorf("could not create event bus: %w", err)
 	}
 
-	eventbus.Benchmark(b, bus)
+	return bus, appID, nil
 }
diff --git a/eventbus/tracing/eventbus_test.go b/eventbus/tracing/eventbus_test.go
index c27312c..ca184cc 100644
--- a/eventbus/tracing/eventbus_test.go
+++ b/eventbus/tracing/eventbus_test.go
@@ -22,6 +22,25 @@ import (
 	"github.com/looplab/eventhorizon/eventbus/local"
 )
 
+// NOTE: Not named "Integration" to enable running with the unit tests.
+func TestAddHandler(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	innerBus := local.NewEventBus()
+	if innerBus == nil {
+		t.Fatal("there should be a bus")
+	}
+
+	bus := NewEventBus(innerBus)
+	if bus == nil {
+		t.Fatal("there should be a bus")
+	}
+
+	eventbus.TestAddHandler(t, bus)
+}
+
 // NOTE: Not named "Integration" to enable running with the unit tests.
 func TestEventBus(t *testing.T) {
 	group := local.NewGroup()