End-to-End Testing Kafka Consumer Groups

1. The Philosophy of End-to-End Testing

End-to-end (E2E) tests are powerful but can be notoriously difficult to manage. They can be slow, fragile, and non-deterministic if not designed carefully. The key is to test our system in a realistic, yet controlled and isolated, environment.

To understand the common pitfalls and the modern approach to this style of testing, let's watch a segment from a video by Dave Farley, a co-author of the book Continuous Delivery.

As the video explains, our goal is to create a deterministic, automated test that treats our broker as a black box (or grey box). We'll control the inputs (produced messages) and verify the outputs (consumed messages and broker state).

This diagram illustrates the flow we are about to test. Our test will orchestrate the Kafka Application Producer and Kafka Application Consumer to validate the behavior of the Embedded Kafka (our broker).

2. Designing the Test Case

A good test tells a story. Here is the story for our E2E test:

1. Arrange:

   • Start a fresh, isolated PostgreSQL database using testcontainers-go.
   • Run our database migrations to set up the schema.
   • Start our broker application, configured to use the test database.
   • Create a test topic, say "logs", with a few partitions.

2. Act:

   • Initialize a producer client and send a batch of unique messages (e.g., 100 messages) to the "logs" topic.
   • Initialize a consumer client, have it join a consumer group (e.g., "log-processors"), and start fetching messages from the "logs" topic.
   • The consumer processes the messages and acknowledges each one.

3. Assert:

   • The consumer must receive the exact number of messages that were sent.
   • The set of received messages must be identical to the set of sent messages (content and order are not guaranteed across partitions, but the set of payloads must match).
   • The crucial check: After the consumer acknowledges the messages, the consumer group's committed offset in the database for each partition should be correct. This confirms that our acknowledgment and offset tracking logic works end-to-end.

This test will validate the interaction between producers, consumers, and consumer groups, which is the heart of our broker's functionality.

3. Implementing the Test Harness

To implement this, we need a "test harness"—a collection of helper functions and infrastructure setup that makes the test itself clean and readable. This aligns with the "custom frameworks" idea from Mitchell Hashimoto's talk on advanced Go testing. We integrate these complex tests directly into the go test toolchain.

Let's build our harness. Create a new file, tests/e2e/main_test.go. We'll put our test logic here.

3.1. Build Tags

First, let's use a build tag. E2E tests are slow and require external dependencies (Docker). We don't want to run them with every file save. Add this line to the top of tests/e2e/main_test.go:

//go:build e2e

package e2e

Now, you can run these tests specifically with go test -v --tags=e2e ./....

3.2. Test Environment Setup with TestMain

We'll use TestMain to set up the database container for all tests in the package. This is much more efficient than starting a container for every single test function. As a refresher, you previously used testcontainers-go for integration tests in Module 4. We are applying the same pattern here.

// tests/e2e/main_test.go

// ... build tag ...

package e2e

import (
	"context"
	"fmt"
	"os"
	"testing"
	"time"

	"github.com/golang-migrate/migrate/v4"
	_ "github.com/golang-migrate/migrate/v4/database/postgres"
	_ "github.com/golang-migrate/migrate/v4/source/file"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/postgres"
	"github.com/testcontainers/testcontainers-go/wait"
)

var dbpool *pgxpool.Pool

func TestMain(m *testing.M) {
	ctx := context.Background()

	// 1. Create a PostgreSQL container
	pgContainer, err := postgres.RunContainer(ctx,
		testcontainers.WithImage("postgres:15-alpine"),
		postgres.WithDatabase("test-db"),
		postgres.WithUsername("user"),
		postgres.WithPassword("password"),
		testcontainers.WithWaitStrategy(
			wait.ForLog("database system is ready to accept connections").
				WithOccurrence(2).
				WithStartupTimeout(5*time.Second),
		),
	)
	if err != nil {
		fmt.Printf("failed to start postgres container: %s\n", err)
		os.Exit(1)
	}
	// Defer container termination
	defer func() {
		if err := pgContainer.Terminate(ctx); err != nil {
			fmt.Printf("failed to terminate postgres container: %s\n", err)
		}
	}()

	// 2. Get connection string and connect
	connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
	if err != nil {
		fmt.Printf("failed to get connection string: %s\n", err)
		os.Exit(1)
	}

	dbpool, err = pgxpool.New(ctx, connStr)
	if err != nil {
		fmt.Printf("failed to connect to database: %s\n", err)
		os.Exit(1)
	}
	defer dbpool.Close()

	// 3. Run migrations
	// Make sure the path to your migrations is correct
	migrationsPath := "file://../../migrations" 
	mig, err := migrate.New(migrationsPath, connStr)
	if err != nil {
		fmt.Printf("failed to create migrate instance: %s\n", err)
		os.Exit(1)
	}
	if err := mig.Up(); err != nil && err != migrate.ErrNoChange {
		fmt.Printf("failed to run migrations: %s\n", err)
		os.Exit(1)
	}

	// 4. Run tests
	exitCode := m.Run()
	os.Exit(exitCode)
}

This setup gives every test function in the file a clean, migrated database connection pool (dbpool) to use.

3.3. Broker and Client Helpers

Inside the same e2e package, we'll create a helpers.go file for functions that start our broker and clients. This follows the principle of creating test helpers that don't return errors but fail the test directly, making the main test logic cleaner.

Here is a conceptual skeleton for a broker helper. You will need to adapt it to your actual server's constructor and configuration struct.

// tests/e2e/helpers.go
//go:build e2e

package e2e

import (
	"context"
	"testing"
	"your-module-path/internal/broker" // Adapt to your project
	"your-module-path/internal/monitoring"
)

// startTestBroker starts a broker instance for testing.
// It returns the running server and a cleanup function.
func startTestBroker(t *testing.T, dbConnectionString string) (*broker.Server, func()) {
	t.Helper()

	// Configure the broker to use the test DB
	config := broker.Config{
		// ... your config fields ...
		DBConnString: dbConnectionString,
	}

	// You'll need to adapt this part to your actual server startup logic
	// For instance, you might not have a separate monitoring struct or registry here.
	// The key is to instantiate and start your server.
	metrics := monitoring.New(nil) // Pass a nil registry if not needed for the test
	server, err := broker.NewServer(config, dbpool, metrics)
	if err != nil {
		t.Fatalf("Failed to create broker server: %v", err)
	}
	
	ctx, cancel := context.WithCancel(context.Background())

	// Run the server in a goroutine
	go func() {
		if err := server.Start(ctx, "localhost:0"); err != nil {
			// Don't fatal here as it might be a clean shutdown
			t.Logf("Broker server stopped: %v", err)
		}
	}()

	// The cleanup function cancels the context, stopping the server.
	cleanup := func() {
		cancel()
	}

	return server, cleanup
}

4. Writing the End-to-End Test

Now we combine the environment, helpers, and test logic into a single, readable test function.

// tests/e2e/main_test.go

func TestProduceConsume_SingleConsumerGroup(t *testing.T) {
	// ARRANGE
	// The DB is already set up by TestMain.
	
	// Start the broker using our helper
	// Note: You need a way to get the actual listener address from the server
	// after it starts, as we used "localhost:0" for a random port.
	brokerServer, cleanup := startTestBroker(t, dbpool.Config().ConnConfig.ConnString())
	defer cleanup()
	
	brokerAddr := brokerServer.GetListenerAddr() // You need to implement this method

	topic := "e2e-logs"
	numPartitions := 4
	
	// Create the topic directly via the broker's logic for simplicity in the test
	err := brokerServer.TopicManager.CreateTopic(context.Background(), topic, numPartitions)
	if err != nil {
		t.Fatalf("Failed to create topic: %v", err)
	}
	
	// ACT
	numMessages := 100
	var sentMessages [][]byte
	
	// Producer client logic (you would wrap this in a helper)
	producer, err := NewProducerClient(brokerAddr) // Your client implementation
	if err != nil {
		t.Fatalf("Failed to create producer: %v", err)
	}
	defer producer.Close()

	for i := 0; i < numMessages; i++ {
		msg := []byte(fmt.Sprintf("message-%d", i))
		sentMessages = append(sentMessages, msg)
		if err := producer.Produce(topic, msg); err != nil {
			t.Fatalf("Failed to produce message: %v", err)
		}
	}

	// Consumer client logic
	consumer, err := NewConsumerClient(brokerAddr, "e2e-group", topic) // Your client implementation
	if err != nil {
		t.Fatalf("Failed to create consumer: %v", err)
	}
	defer consumer.Close()

	receivedMessagesChan := make(chan []byte, numMessages)
	go func() {
		for i := 0; i < numMessages; i++ {
			msg, err := consumer.Consume(context.Background())
			if err != nil {
				t.Errorf("Failed to consume message: %v", err)
				return
			}
			receivedMessagesChan <- msg.Payload
			if err := consumer.Ack(msg); err != nil {
				t.Errorf("Failed to ack message: %v", err)
			}
		}
	}()

	// ASSERT
	receivedCount := 0
	receivedMessages := make(map[string]bool)
	
	// Wait for all messages or timeout
	timeout := time.After(10 * time.Second)
	for receivedCount < numMessages {
		select {
		case msg := <-receivedMessagesChan:
			receivedCount++
			receivedMessages[string(msg)] = true
		case <-timeout:
			t.Fatalf("Timed out waiting for messages. Received %d of %d", receivedCount, numMessages)
		}
	}
	
	// Assert count
	if receivedCount != numMessages {
		t.Errorf("Expected to receive %d messages, but got %d", numMessages, receivedCount)
	}

	// Assert message content
	for _, sentMsg := range sentMessages {
		if !receivedMessages[string(sentMsg)] {
			t.Errorf("Did not receive expected message: %s", string(sentMsg))
		}
	}
	
	// Assert final committed offset (grey-box testing)
	// You might need a small delay to ensure the last ack is processed
	time.Sleep(100 * time.Millisecond) 
	
	// Logic to query your DB and check the committed offset for the group.
	// For example:
	// finalOffset, err := getCommittedOffset(dbpool, "e2e-group", topic, partitionID)
	// require.NoError(t, err)
	// require.Equal(t, expectedFinalOffset, finalOffset)
}

This test is comprehensive. It validates not just the data path but also the critical state management of consumer group offsets. For an example of a real-world project that uses a similar testing strategy, you can look at Splunk's kafka-mq-go repository.

Conclusion

Congratulations! You have just designed and implemented a professional-grade end-to-end test for a complex, stateful, networked application. This is a significant milestone that moves your project much closer to being production-ready.

Key Takeaways:

• Isolated E2E Testing: You've adopted the modern philosophy of creating focused, deterministic acceptance tests that validate your system without the brittleness of testing an entire integrated environment.
• Test Harnesses: You've used TestMain, build tags, and helper functions to create a clean and maintainable test suite for complex scenarios.
• Grey-Box Validation: By asserting both the external behavior (messages received) and the internal state (database offsets), you've created a test with a very high degree of confidence.

Next Up

So far, we have used containers as a dependency for our tests. In the next and final lesson of the course, we will flip this around and use containers to package and deploy our application. We will containerize the broker and use Docker Compose to orchestrate its local deployment, bringing all the pieces of our project together.