Forem: Jose Boretto

How to join multiple Kafka topics - Benchmark

Jose Boretto — Sat, 19 Oct 2024 10:45:27 +0000

Approach comparison

Apache Kafka is a powerful tool for building scalable real-time streaming applications, and a common use case is joining data from multiple Kafka topics. Depending on the needs of your application, there are several ways to approach this problem. In this article, we'll explore three approaches:

N consumers write to 1 table (reader-oriented) using Kafka consumers.
N consumers write to N tables (writer-oriented) using Kafka consumers.
N consumers write to 1 table using Kafka Streams.

Each of these strategies has its own trade-offs in terms of complexity, scalability, and performance. If you want to deep dive in each alternative, please check the following post: https://dev.to/joseboretto/how-to-join-multiple-kafka-topics-overview-19j8 .

Feature request

What do we need to satisfy our user's needs?

Allow our sellers to see and filter their catalog in https://www.manomano.fr/
Allow our sellers to see their catalog stats in https://www.manomano.fr/

System load

The microservice has to manage the following load.

Kafka Throughput (Writer)


Topic	Per Day (Million)	Requests/s	Partitions
inboundoffer	27.3	316.2	6
offer	24.62	284.9	6
product	13.13	152	12
availablestock	3.2	37	6
xcart_products	1.26	14	4
colibri_product_generic	0.13	1	4

Database size (Reader)

N Consumers Write to N Tables (Writer-Oriented)


Table	Rows (Million)	Size (GB)	Avg Size (Bytes)
inboundoffer	96	189	1968
offer	107	43	2488
product	126	90	1400
availablestock	92	10.2	108
xcart_products	118	32	271
colibri_product_generic	92	9.5	103

N consumers write to 1 table using Kafka Streams.


Table	Rows (Million)	Size (GB)	Avg Size (Bytes)
main	119	147	1235

Platform infrastructure

Database

Provider: Amazon RDS - https://aws.amazon.com/rds/
Engine: PostgreSQL
Size: db.r6g.8xlarge
Config: 2 readers and 1 writer


Model	Core Count	vCPU	Memory (GiB)	Storage	Dedicated EBS Bandwidth (Mbps)	Networking Performance (Gbps)
db.r6g.large	-	2	16	EBS-Only	up to 4,750	Up to 10
db.r6g.xlarge	-	4	32	EBS-Only	up to 4,750	Up to 10
db.r6g.2xlarge	-	8	64	EBS-Only	up to 4,750	Up to 10
db.r6g.4xlarge	-	16	128	EBS-Only	4,750	Up to 10
db.r6g.8xlarge	-	32	256	EBS-Only	9,000	12
db.r6g.12xlarge	-	48	384	EBS-Only	13,500	20
db.r6g.16xlarge	-	64	512	EBS-Only	19,000	25

Kafka cluster

Provider: Amazon MSK - https://aws.amazon.com/msk/
Version: 3.6
Brokers: 6
Size: kafka.m5.4xlarge


Broker Instance	vCPU	Memory (GiB)
kafka.t3.small	2	2
kafka.m5.large	2	8
kafka.m5.xlarge	4	16
kafka.m5.2xlarge	8	32
kafka.m5.4xlarge	16	64
kafka.m5.8xlarge	32	128
kafka.m5.12xlarge	48	192
kafka.m5.16xlarge	64	256
kafka.m5.24xlarge	96	384

Case 2: N Consumers Write to N Tables (Writer-Oriented) with Kafka Consumer

In this method, each consumer reads from its Kafka topic and writes to its own dedicated table in the database. This is a "writer-oriented" approach, where each consumer is responsible for maintaining its isolated dataset.

Hardware

Provider: Amazon Elastic Kubernetes Service (EKS)
Pods: 6
Memory: 8GiB (Used: 50%)
CPU: 2000m (Used: 100%)

Software

Language: Java 21
Framework:
- Spring Boot 3.2.2
- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot/3.2.2
- Reactor Core 3.6.4
- https://mvnrepository.com/artifact/io.projectreactor/reactor-core/3.6.4

Performance

Writer


Resource Name	Requests	Avg Latency	p50 Latency	p75 Latency	p90 Latency	p95 Latency	Requests/s
inboundOffer	30.7M	94.2 μs	79.3 μs	95.5 μs	126 μs	162 μs	355 req/s
offer	26.1M	190 μs	157 μs	224 μs	283 μs	310 μs	302 req/s
product	10.8M	431 μs	380 μs	450 μs	595 μs	717 μs	125 req/s
availableStock	2.58M	182 μs	178 μs	214 μs	258 μs	283 μs	29.9 req/s
xcart_products	1.27M	95.2 μs	62.9 μs	124 μs	192 μs	221 μs	14.6 req/s
colibri_product_generic	129k	74.1 μs	54.7 μs	76.9 μs	132 μs	170 μs	1.5 req/s

Reader


Resource Name	Requests	Avg Latency	p50 Latency	p75 Latency	p90 Latency	p95 Latency	Requests/s
GET products v1	100.0k	1.01 s	77.4 ms	413 ms	925 ms	1.89 s	0.2 req/s
GET stats v1	27.0k	5.76 s	371 ms	2.17 s	10.4 s	26.8 s	< 0.1 req/s

Database Considerations:

This architecture requires joining 6 tables.
All indexes are optimized for search by filters.
No partitioned tables.

Case 3: Kafka Streams Write to Another Kafka Topic, then a Consumer Writes to 1 table

In this enhanced version of the Kafka Streams approach, Kafka Streams instances read from multiple topics, process the data (joins, aggregations, filtering, etc.), and then write the processed output to a new Kafka topic. A separate Kafka consumer (or consumers) can then read from this topic and handle the writes to the database.

Hardware

Provider: Amazon Elastic Kubernetes Service (EKS)
Pods: 12
Memory: 8GiB (Used: 100%)
CPU: 4000m (Used: 20%)
Disk: 100Gi (Used: 10%)
Kafka internal topics: 26 topics with total size of 500G (without taking into account the replication factor)

Software

Language: Java 21
Framework:
- Spring Boot 3.2.2
- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot/3.2.2
- Kafka Streams 3.6.2
- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams/3.6.2

Performance

Kafka stream writer (From input topic to main topic)


Resource Name	Requests	Avg Latency	p50 Latency	p75 Latency	p90 Latency	p95 Latency	Requests/s
inboundOffer	61.0M	21.5 μs	15.1 μs	21.9 μs	44.0 μs	62.9 μs	706 req/s
offer	51.8M	59.1 μs	31.3 μs	76.9 μs	134 μs	183 μs	599 req/s
product	21.9M	305 μs	92.6 μs	392 μs	740 μs	1.07 ms	254 req/s
availableStock	5.17M	50.6 μs	34.9 μs	67.9 μs	103 μs	126 μs	59.9 req/s
xcart_products	2.47M	33.7 μs	17.4 μs	38.9 μs	89.8 μs	122 μs	28.6 req/s
colibri_product_generic	259k	26.9 μs	13.3 μs	32.8 μs	60.9 μs	84.4 μs	3.0 req/s

Kafka writer (From main topic to database)


Resource Name	Requests	Avg Latency	p50 Latency	p75 Latency	p90 Latency	p95 Latency	Requests/s
Main topic	17.2M	17.9 μs	12.5 μs	23.7 μs	32.3 μs	36.0 μs	199 req/s

Reader


Resource Name	Requests	Avg Latency	p50 Latency	p75 Latency	p90 Latency	p95 Latency	Requests/s
GET products v2	49.7k	211 ms	44.3 ms	114 ms	365 ms	756 ms	< 0.1 req/s
GET stats v2	9.12k	135 ms	34.6 ms	114 ms	308 ms	599 ms	< 0.1 req/s

Database Considerations:

Queries involve only 1 table (no joins).
Partitioned table with all necessary indexes for filtering.

Performance Comparison


Metric	N Consumers Write to N Tables (Writer-Oriented)	Kafka Streams
Write Throughput (requests/s)
InboundOffer	355 req/s	706 req/s
Offer	302 req/s	599 req/s
Product	125 req/s	254 req/s
AvailableStock	29.9 req/s	59.9 req/s
Xcart_Products	14.6 req/s	28.6 req/s
Colibri_Product_Generic	1.5 req/s	3.0 req/s
p50 Write Latency
InboundOffer	79.3 μs	15.1 μs
Offer	157 μs	31.3 μs
Product	380 μs	92.6 μs
AvailableStock	178 μs	34.9 μs
Xcart_Products	62.9 μs	17.4 μs
Colibri_Product_Generic	54.7 μs	13.3 μs
GET Products
p50 Read Latency	77.4 ms	44.3 ms
p90 Read Latency	925 ms	365 ms
GET Stats
p50 Read Latency	371 ms	34.6 ms
p90 Read Latency	10.4 s	308 ms
Infrastructure
CPU Usage	100% CPU usage (on 2000m CPU, 6 pods)	20% CPU usage (on 4000m CPU, 12 pods)
Memory Usage	50% of 8GiB (6 pods)	100% of 8GiB (12 pods)
Disk Usage	0 GiB	10% of 100GiB (12 pods)
Database Write Complexity	Writes to 6 tables, requires joining for reads	Writes to 1 table, no joins required for reads

Conclusion

Write Latency (p50): Kafka Streams achieves much lower p50 write latency across all topics, indicating faster processing of the majority of requests.
Read Latency (p50): Kafka Streams shows a clear advantage in read latency, with a p50 latency of 44.3 ms compared to 77.4 ms in GET Products (2x improvement) and p50 latency of 34.6 ms compared to 371 ms in GET Stats (10x improvement) for the N Consumers approach. This improvement is due to writing to a single table, avoiding complex joins.
Resource Efficiency: Kafka Streams uses more memory and disk space.

Credits

To all the developers involved in this 3 year journey.

How to join multiple Kafka topics - Overview

Jose Boretto — Sat, 19 Oct 2024 10:25:50 +0000

Approach comparison

N consumers write to 1 table (reader-oriented) using Kafka consumers.
N consumers write to N tables (writer-oriented) using Kafka consumers.
N consumers write to 1 table using Kafka Streams.

Each of these strategies has its own trade-offs in terms of complexity, scalability, and performance. Let's analyze each one in detail.

1. N Consumers Write to One Table (Reader-Oriented) with Kafka Consumer

In this approach, multiple Kafka consumers are each responsible for reading from their respective Kafka topics and writing to a single database table. This method assumes that all consumers are reading from distinct topics, and once they consume data, they normalize or aggregate the data into a unified format and insert it into a single table.

Advantages:

Simplicity: This is a straightforward approach to implementation. Each consumer focuses only on reading from its designated topic and writing to the same target database.
Single View: A single table in the database provides a unified view of data from all topics, making queries easier for end-users or downstream systems.

Challenges:

Concurrency Issues: With multiple consumers writing to the same table, there’s a potential for race conditions, deadlocks, or row-level locking, especially in databases that don't handle concurrent writes well.
Scalability: As the volume of data or the number of topics grows, contention for database resources increases, affecting write performance.
Transformation Logic: All consumers must implement consistent transformation logic to maintain data integrity, which can increase complexity if the topics have varying schemas or structures.
Write Bottlenecks: Since all writes funnel into a single table, database performance could become a bottleneck.

N+1 Problem:

This approach can encounter the N+1 problem when a single event needs to trigger writes to multiple rows in the database. For example, if one event results in N separate rows being written (or updated), the system must execute N database transactions for each event. This can lead to performance degradation as the number of rows grows, significantly slowing down writes, and increasing database load.

Performance Impact: The N+1 problem can cause severe performance bottlenecks, especially in high-throughput scenarios. The larger N becomes, the more time and resources are required to process a single event, which could overwhelm the database with excessive writes.

When to Use:

You need a unified view of data and the topics have similar structures.
Simplicity is key, and your database can handle concurrent writes at scale.

2. N Consumers Write to N Tables (Writer-Oriented) with Kafka Consumer

Advantages:

Decoupling: Each consumer is entirely independent of the others. There are no shared resources, so there's no risk of write contention.
High Performance: Since each consumer has its own table, there are no race conditions, locking issues, or competition for table-level resources in the database. This leads to better performance when dealing with large volumes of data.
Easier Debugging: If one consumer or table encounters a problem, it does not affect the others, making it easier to isolate and fix issues.

Challenges:

Querying: Since data is spread across multiple tables, querying becomes more complex. If you need a unified view of the data, you would need to perform cross-table joins, which can degrade query performance.
Data Consistency: Ensuring consistency across tables can be tricky, especially if tables are updated out of sync or at different rates.

When to Use:

High data volume where write performance is critical.
The data in the topics is diverse or unrelated, so there’s no need for a unified view in a single table.
You want clear separation of concerns for each data stream.

3. Kafka Streams Write to Another Kafka Topic, then a Consumer Writes to 1 table

Advantages:

Decoupling Stream Processing and Database Writes: By separating the stream processing from the database writes, you prevent the database from becoming a bottleneck. Kafka Streams focuses solely on processing and transforming data, while a separate consumer manages the database interaction.
Scalability: Both the Kafka Streams application and the consumer that writes to the database can scale independently. For instance, if database writes become slow, you can scale the consumer group writing to the database without affecting the Kafka Streams application.
Backpressure Handling: Kafka, with its persistent log structure, can act as a buffer between the stream processing and the database writes. If the database slows down, Kafka retains the processed data in the intermediate topic, allowing the system to recover without losing data.
Fault Tolerance: Kafka Streams offers built-in fault tolerance with state stores and commits to the Kafka topic, while the downstream consumer group writing to the database can be monitored separately for failures, making the entire pipeline more resilient.

Challenges:

Learning Curve: Kafka Streams has a steeper learning curve compared to basic Kafka consumers, especially for teams unfamiliar with the stream processing paradigm.
Internal Complexity: Kafka Streams introduces a higher level of abstraction, such as KTable, KStream, KStore, and Topology, which can make debugging difficult. Understanding how states are managed (using RocksDB) and how the stream’s topology is built and executed adds complexity. If issues arise in how data is persisted or transformed, tracking down the root cause across these abstractions requires deep familiarity with Kafka Streams internals.
Debugging State Stores: Investigating state store corruption, lag, or rebuilds, especially with complex stateful operations, can be challenging.
RocksDB Management: Kafka Streams uses RocksDB for local storage, which can introduce performance issues if not tuned correctly.
Topology Visualization: Understanding the full topology (data flow, partitioning, transformations) can be hard without proper tooling, making the system more opaque.
Cost: Kafka Streams can be expensive due to its use of local state stores, changelog topics, and repartitioning, which increase disk usage, network traffic, and Kafka storage overhead. High CPU and memory consumption arise from complex stream operations and large in-memory buffers.
Kafka events keys: All the events must have a key.

When to Use:

When stream processing and database writes require decoupling for scalability and reliability.
When you want to prevent the database from being overwhelmed by a sudden surge in data.
When you want to take advantage of Kafka’s backpressure and buffering capabilities to absorb variations in database write performance.

Learnings from our experience

Over the past three years, we've gained significant insights into optimizing Kafka-based data pipelines, especially in scenarios involving high throughput and complex data processing. Here are the key lessons we’ve learned:

1- Process Events in Batches to Reduce Database Commits:

By processing events in batches and inserting them in bulk, we’ve reduced the number of commits to the database, which greatly improves performance. Batching minimizes the overhead of frequent database writes and allows more efficient resource usage.

2- Fine-Tuning Kafka Consumers:

Optimizing consumer configurations, such as max.poll.records (which controls how many records a consumer can fetch in a single call) and max.partition.fetch.bytes (which determines the maximum size of data a consumer can fetch per partition), significantly enhanced our ability to manage large data volumes without overwhelming consumers.

3- Dual Database Connections:

We learned that separating database connections for writing and reading improves overall performance. A dedicated writer connection handles inserts and updates efficiently, while a reader connection optimizes read-heavy operations without contention for database resources.

4- Discard Duplicate Events:

Handling duplicate events is crucial for maintaining data consistency and reducing unnecessary processing of both the input and the output topic. We implemented deduplication strategies to discard redundant events, preventing potential performance hits from processing the same data multiple times.

5- Using Partitioned Tables:

Partitioning large database tables based on specific keys (such as user_id) improved query performance and simplified data management. Partitioned tables allowed faster reads and writes, reducing the load on individual table partitions.

6- Leverage Reactive Programming:

By adopting reactive programming techniques, we were able to make full use of the CPU resources in our microservices. Reactive code enabled us to handle high volumes of asynchronous events efficiently, improving overall throughput without overloading the system.

7- Upserts for Efficient Data Storage:

Instead of performing a SELECT followed by INSERT or UPDATE, we switched to using upserts, which directly insert new records or update existing ones in a single transaction. This eliminated race conditions and improved data processing speed.

8- Dead Letter Queue (DLQ) for Each Topic:

Implementing a DLQ for each Kafka topic ensures that when the database throws an error, you will not lose data. This allowed us to handle errors effectively, preventing data loss and enabling reprocessing of failed events as needed.

These practices have proven invaluable in ensuring the efficiency, scalability, and reliability of our Kafka-based systems, enabling us to handle increasing data volumes while maintaining high performance and data integrity.

Conclusion

Using Kafka Streams to process data and writing the results to an intermediate Kafka topic before a separate consumer writes to the database is an optimal approach for real-time systems that require high scalability and fault tolerance. This method decouples stream processing from database interactions, preventing the database from becoming a bottleneck, while providing Kafka’s native backpressure and scalability benefits.

It is particularly useful when dealing with complex joins and aggregations across multiple topics, as Kafka Streams excels at processing and transforming data in a reliable, fault-tolerant manner.

Benchmark

https://dev.to/joseboretto/how-to-join-multiple-kafka-topics-benchmark-2heb

Credits

To all the developers involved in this 3 year journey.

Integration Tests in Go with Cucumber, Testcontainers, and HTTPMock

Jose Boretto — Wed, 11 Sep 2024 22:57:00 +0000

Integration Testing in Go with Cucumber, Testcontainers, and HTTPMock

Integration testing is a crucial part of the software development lifecycle that ensures different modules of an
application work seamlessly together. In the Go (Golang) ecosystem, we can use various tools and libraries to facilitate
integration testing. This post will explore how to use Cucumber, Testcontainers for our PostgreSQL database, and
HTTPMock to write robust integration tests in Go.

Why Integration Testing Matters

Integration testing verifies the interactions between different parts of your application, such as external services,
databases, and APIs. Unlike unit tests, which test individual components in isolation, integration tests ensure the
entire system functions as intended when combined.

Tools Overview

Cucumber: A tool that supports Behavior-Driven Development (BDD). It allows you to write human-readable test scenarios in the Gherkin language, which bridges the gap between technical and non-technical stakeholders.
1. https://github.com/cucumber/godog
Testcontainers: A library that provides lightweight, disposable containers for testing. It allows you to spin up real instances of databases, message brokers, or any other service your application depends on, making your tests more reliable and closer to real-world scenarios. We will use it to start our PostgresSQL database.
1. https://github.com/testcontainers/testcontainers-go
HTTPMock: A simple HTTP mocking library that allows you to simulate HTTP interactions in your tests. It is useful when you want to mock responses from external HTTP services without actually making network requests.
1. https://github.com/jarcoal/httpmock

Use case

We will have 1 application that it’s in charge of executing the following logic:

Implement an API to create a book.
Check if the ISBN of the book exist in a third party API.
Store the book in the database
Send and email using a third-party API to the address helloworld@gmail.com

Integration test architecture

We will treat our application as a black box. This means we have to start the application and use it as any other
user, without mocking any part of the code.

The folder structure of project will be

- cmd/  
   - features/createBook.feature  
   - testAssets/testData/dev-db.sql
   - integration_test.go  
   - main.go  
   - step_definition_api.go  
   - step_definition_common.go  
   - step_definition_database.go  
   - step_definition_mock_server.go  
   - test_utils.go  
   - testcontainers_config.go  
- go.mod

Step-by-Step Guide to Integration Testing in Go

Let’s walk through the process of setting up integration tests in Go using Cucumber, Testcontainers, and HTTPMock.

1. Installing Required Libraries

Next, install the necessary libraries

go get -u github.com/cucumber/godog  
go get -u github.com/testcontainers/testcontainers-go  
go get -u github.com/jarcoal/httpmock  
go get -u github.com/lib/pq  
go get -u gorm.io/driver/postgres

2. Update your main.go to control the execution fo the app

Now, we have to change a bit our main.go. Separate your main in 3 different functions.

func mainHttpServerSetup(addr string, httpClient *http.Client) (*http.Server, func())
- This function is used in testcontainers_config.go allowing us control the execution of the app in integration_test.go
main()
- The entry point of the actual application.

package main

import (
    "fmt"
    servicebook "github.com/joseboretto/golang-testcontainers-gherkin-setup/internal/application/services/books"
    controller "github.com/joseboretto/golang-testcontainers-gherkin-setup/internal/infrastructure/controllers"
    controllerbook "github.com/joseboretto/golang-testcontainers-gherkin-setup/internal/infrastructure/controllers/books"
    "gorm.io/driver/postgres"
    "gorm.io/gorm"
    "gorm.io/gorm/schema"
    "os"
    "time"

    "log"
    "net/http"

    clientsbook "github.com/joseboretto/golang-testcontainers-gherkin-setup/internal/infrastructure/clients/book"
    persistancebook "github.com/joseboretto/golang-testcontainers-gherkin-setup/internal/infrastructure/persistance/book"
)

func main() {
    addr := ":8000"
    httpClient := &http.Client{
        Timeout: 5 * time.Second,
    }
    server, deferFn := mainHttpServerSetup(addr, httpClient)
    log.Println("Listing for requests at http://localhost" + addr)
    err := server.ListenAndServe()
    if err != nil {
        panic("Error staring the server: " + err.Error())
    }
    defer deferFn()
}

func mainHttpServerSetup(addr string, httpClient *http.Client) (*http.Server, func()) {
    db := getDatabaseConnection()
    // Migrate the schema
    err := db.AutoMigrate(&persistancebook.BookEntity{})
    if err != nil {
        panic("Error executing db.AutoMigrate" + err.Error())
    }
    // Clients
    checkIsbnClientHost := os.Getenv("CHECK_ISBN_CLIENT_HOST")
    checkIsbnClient := clientsbook.NewCheckIsbnClient(checkIsbnClientHost, httpClient)
    emailClientHost := os.Getenv("EMAIL_CLIENT_HOST")
    sendEmailClient := clientsbook.NewSendEmailClient(emailClientHost, httpClient)
    // repositories
    newCreateBookRepository := persistancebook.NewCreateBookRepository(db)
    // services
    createBookService := servicebook.NewCreateBookService(newCreateBookRepository, checkIsbnClient, sendEmailClient)
    // controllers
    bookController := controllerbook.NewBookController(createBookService)
    // routes
    controller.SetupRoutes(bookController)
    // Server
    server := &http.Server{Addr: addr, Handler: nil}
    // Defer function
    // Add all defer
    deferFn := func() {
        fmt.Println("closing database connection")
        sqlDB, err := db.DB()
        err = sqlDB.Close()
        if err != nil {
            panic("Error closing database connection: " + err.Error())
        }

    }
    return server, deferFn
}

func getDatabaseConnection() *gorm.DB {
    // Read database configuration from environment variables
    databaseUser := os.Getenv("DATABASE_USER")
    databasePassword := os.Getenv("DATABASE_PASSWORD")
    databaseName := os.Getenv("DATABASE_NAME")
    databaseHost := os.Getenv("DATABASE_HOST")
    databasePort := os.Getenv("DATABASE_PORT")
    databaseConnectionString := `host=` + databaseHost + ` user=` + databaseUser + ` password=` + databasePassword + ` dbname=` + databaseName + ` port=` + databasePort + ` sslmode=disable`
    fmt.Println("databaseConnectionString: ", databaseConnectionString)
    db, err := gorm.Open(postgres.New(postgres.Config{
        DSN:                  databaseConnectionString,
        PreferSimpleProtocol: true, // disables implicit prepared statement usage
    }), &gorm.Config{
        NamingStrategy: schema.NamingStrategy{
            TablePrefix:   "myschema.", // schema name
            SingularTable: false,
        }})
    // Check if connection is successful
    if err != nil {
        panic("failed to connect database with error: " + err.Error() + "\n" + "Please check your database configuration: " + databaseConnectionString)
    }
    return db
}

3. Writing the Feature File (Cucumber)

Create a feature file (createBook.feature) inside the features directory with scenarios that describe the expected
behavior of your application.


Feature: Create book

   Background: Clean database
      Given SQL command
    """
    DELETE FROM myschema.books;
    """
      And reset mock server

   Scenario: Create a new book successfully
      Given a mock server request with method: "GET" and url: "https://api.isbncheck.com/isbn/0-061-96436-1"
      And a mock server response with status 200 and body
    """json
    {
       "id": "0-061-96436-1"
    }
    """
      And a mock server request with method: "POST" and url: "https://api.gmail.com/send-email" and body
    """json
    {
      "email" : "helloworld@gmail.com",
      "book" : {
        "isbn" : "0-061-96436-1",
        "title" : "The Art of Computer Programming"
      }
    }
    """
      And a mock server response with status 200 and body
    """json
    {
       "status": "OK"
    }
    """
      When API "POST" request is sent to "/api/v1/createBook" with payload
    """json
    {
      "isbn": "0-061-96436-1",
      "title": "The Art of Computer Programming"
    }
    """
      Then API response status code is 200 and payload is
    """json
    {
        "isbn": "0-061-96436-1",
        "title": "The Art of Computer Programming"
    }
    """
      And SQL query "SELECT * FROM myschema.books WHERE isbn = '0-061-96436-1'" result without the fields "created_at,deleted_at,updated_at" is equal to
    """json
    [
       {
          "id":1,
          "isbn":"0-061-96436-1",
          "title":"The Art of Computer Programming"
       }
    ]
    """

4. Implementing Step Definitions in Go

Now, create the Go files to implement the step definitions for the scenarios. We have multiple step definitions based on
the goals.

The most important concept to remember is writing generic step definitions to re-use them on every use case.

step_definition_common.go

Share data between the app setup and the step definition.

package main

import (
    "database/sql"
    "github.com/cucumber/godog"
    "net/http"
)

type StepsContext struct {
    // Main setup
    mainHttpServerUrl string // http://localhost:8000
    database          *sql.DB
    // Mock server setup
    stepMockServerRequestMethod *string
    stepMockServerRequestUrl    *string
    stepMockServerRequestBody   *string
    stepResponse                *http.Response
}

func NewStepsContext(mainHttpServerUrl string, database *sql.DB, sc *godog.ScenarioContext) *StepsContext {
    s := &StepsContext{
        mainHttpServerUrl: mainHttpServerUrl,
        database:          database,
    }
    // Register all the step definition function
    s.RegisterMockServerSteps(sc)
    s.RegisterDatabaseSteps(sc)
    s.RegisterApiSteps(sc)
    return s
}

step_definition_api.go

Perform api request and check the response.

package main

import (
    "bytes"
    "context"
    "fmt"
    "github.com/cucumber/godog"
    "io"
    "log"
    "net/http"
    "time"
)

func (s *StepsContext) RegisterApiSteps(sc *godog.ScenarioContext) {
    sc.Step(`^API "([^"]*)" request is sent to "([^"]*)" without payload$`, s.apiRequestIsSendWithoutPayload)
    sc.Step(`^API "([^"]*)" request is sent to "([^"]*)" with payload$`, s.apiRequestIsSendWithPayload)
    sc.Step(`^API response status code is (\d+) and payload is$`, s.apiResponseIs)
}

func (s *StepsContext) apiRequestIsSendWithoutPayload(method, url string) error {
    return s.apiRequestIsSendWithPayload(method, url, "")
}

func (s *StepsContext) apiRequestIsSendWithPayload(method, url, payloadJson string) error {
    client := &http.Client{
        Timeout: 5 * time.Second,
    }

    req, err := http.NewRequestWithContext(context.Background(), method, s.mainHttpServerUrl+url, bytes.NewBufferString(payloadJson))
    if err != nil {
        return fmt.Errorf("failed to create a new HTTP request: %w", err)
    }

    req.Header.Set("Content-Type", "application/json")

    response, err := client.Do(req)
    if err != nil {
        return fmt.Errorf("failed to execute HTTP request: %w", err)
    }

    s.stepResponse = response
    return nil
}

func (s *StepsContext) apiResponseIs(expected int, expectedResponse string) error {
    defer s.stepResponse.Body.Close()

    if s.stepResponse.StatusCode != expected {
        return fmt.Errorf("expected status code %d but got %d", expected, s.stepResponse.StatusCode)
    }

    actualJson, err := getBody(s.stepResponse)
    if err != nil {
        return err
    }

    if match, err := compareJSON(expectedResponse, actualJson); err != nil {
        return fmt.Errorf("error comparing JSON: %w", err)
    } else if !match {
        log.Printf("Actual response body: %s", actualJson)
        return fmt.Errorf("response body does not match. Expected: %s, actual: %s", expectedResponse, actualJson)
    }

    return nil
}

func getBody(response *http.Response) (string, error) {
    body, err := io.ReadAll(response.Body)
    if err != nil {
        return "", fmt.Errorf("failed to read response body: %w", err)
    }
    return string(body), nil
}

step_definition_database.go

store data in the database. Additionally, you have to implement the step definitions to check the data stored in the
database.

package main

import (
    "encoding/json"
    "fmt"
    "github.com/cucumber/godog"
    "strings"
)

func (s *StepsContext) RegisterDatabaseSteps(sc *godog.ScenarioContext) {
    sc.Step(`^SQL command`, s.executeSQL)
    sc.Step(`^SQL query "([^"]*)" result is equal to`, s.checkSQLqueryWithoutIgnore)
    sc.Step(`^SQL query "([^"]*)" result without the fields "([^"]*)" is equal to`, s.checkSQLqueryWithIgnoredFields)
}

func (s *StepsContext) executeSQL(sqlCommand string) error {
    _, err := s.database.Exec(sqlCommand)
    if err != nil {
        return err
    }
    return nil
}

func (s *StepsContext) checkSQLqueryWithoutIgnore(query, jsonString string) error {
    return s.checkSQLqueryWithIgnoredFields(query, jsonString, "")
}

func (s *StepsContext) checkSQLqueryWithIgnoredFields(query, ignoredFields, jsonString string) error {
    // Parse ignored fields into a map for quick lookup
    ignoredFieldsSet := make(map[string]struct{})
    if ignoredFields != "" {
        for _, field := range strings.Split(ignoredFields, ",") {
            ignoredFieldsSet[strings.TrimSpace(field)] = struct{}{}
        }
    }
    // Execute the SQL query
    rows, err := s.database.Query(query)
    if err != nil {
        return fmt.Errorf("error executing query: %w", err)
    }
    defer rows.Close()

    // Fetch column names
    columns, err := rows.Columns()
    if err != nil {
        return fmt.Errorf("error fetching columns: %w", err)
    }

    // Prepare a slice of maps to hold query results
    var resultRows []map[string]interface{}

    for rows.Next() {
        // Create a slice to hold column values
        columnValues := make([]interface{}, len(columns))
        columnPointers := make([]interface{}, len(columns))

        for i := range columnValues {
            columnPointers[i] = &columnValues[i]
        }

        // Scan the result into column pointers
        if err := rows.Scan(columnPointers...); err != nil {
            return fmt.Errorf("error scanning row: %w", err)
        }

        // Create a map to represent a row
        rowMap := make(map[string]interface{})
        for i, colName := range columns {
            val := columnValues[i]

            // Convert bytes to string for easier comparison
            if b, ok := val.([]byte); ok {
                rowMap[colName] = string(b)
            } else {
                rowMap[colName] = val
            }
        }

        resultRows = append(resultRows, rowMap)
    }
    // Remove ignored fields from the actual data
    for i := range resultRows {
        for ignoredField := range ignoredFieldsSet {
            delete(resultRows[i], ignoredField)
        }
    }
    // Remove ignored fields from the expected data
    var expectedData []map[string]interface{}
    if err := json.Unmarshal([]byte(jsonString), &expectedData); err != nil {
        return fmt.Errorf("error unmarshalling provided JSON string: %w", err)
    }
    for i := range expectedData {
        for ignoredField := range ignoredFieldsSet {
            delete(expectedData[i], ignoredField)
        }
    }
    // Convert query result to JSON
    queryResultJSON, err := json.Marshal(resultRows)
    if err != nil {
        return fmt.Errorf("error marshalling query result to JSON: %w", err)
    }
    //
    if match, err := compareJSONArrays(jsonString, string(queryResultJSON)); err != nil {
        return fmt.Errorf("error comparing JSON: %w", err)
    } else if !match {

        return fmt.Errorf("actual Expected: %s, actual: \n %s", jsonString, string(queryResultJSON))
    }
    return nil
}

test_utils.go

Mainly, this file contains the function to compare JSON strings.


package main

import (
   "encoding/json"
   "fmt"
   "reflect"
)

// compareJSON compares two JSON strings and returns true if they are equal.
func compareJSON(jsonStr1, jsonStr2 string) (bool, error) {
   var obj1, obj2 map[string]interface{}

   // Unmarshal the first JSON string
   if err := json.Unmarshal([]byte(jsonStr1), &obj1); err != nil {
      return false, fmt.Errorf("error unmarshalling jsonStr1: %v", err)
   }

   // Unmarshal the second JSON string
   if err := json.Unmarshal([]byte(jsonStr2), &obj2); err != nil {
      return false, fmt.Errorf("error unmarshalling jsonStr2: %v", err)
   }
   // Compare the two maps
   return reflect.DeepEqual(obj1, obj2), nil
}

func compareJSONArrays(jsonStr1, jsonStr2 string) (bool, error) {
   var obj1, obj2 []map[string]interface{}

   // Unmarshal the first JSON string
   if err := json.Unmarshal([]byte(jsonStr1), &obj1); err != nil {
      return false, fmt.Errorf("error unmarshalling jsonStr1: %v", err)
   }

   // Unmarshal the second JSON string
   if err := json.Unmarshal([]byte(jsonStr2), &obj2); err != nil {
      return false, fmt.Errorf("error unmarshalling jsonStr2: %v", err)
   }
   // Compare the two maps
   return reflect.DeepEqual(obj1, obj2), nil
}

5. Configure test containers

testcontainers_config.go

Configure TestContainersParams to match the envs and the config of you actual application. Additionally, httpmock.ActivateNonDefault(mockClient) is key ot make httpmock works because it will only mock our third
party http.client

package main

import (
    "context"
    "database/sql"
    "fmt"
    "github.com/jarcoal/httpmock"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/postgres"
    "log"
    "net/http"
    "os"
    "path/filepath"
    "time"

    "github.com/docker/go-connections/nat"

    _ "github.com/lib/pq" // Import the postgres driver
    "github.com/testcontainers/testcontainers-go/wait"
)

type TestContainersParams struct {
    MainHttpServerAddress  string
    PostgresImage          string
    DatabaseName           string
    DatabaseHostEnvVar     string
    DatabasePortEnvVar     string
    DatabaseUserEnvVar     string
    DatabasePasswordEnvVar string
    DatabaseNameEnvVar     string
    DatabaseInitScript     string
    EnvironmentVariables   map[string]string
}

type TestContainersContext struct {
    MainHttpServer *http.Server
    Database       *sql.DB
    Params         *TestContainersParams
}

func NewTestContainersParams() *TestContainersParams {
    return &TestContainersParams{
        MainHttpServerAddress:  ":8000",
        PostgresImage:          "docker.io/postgres:16-alpine",
        DatabaseName:           "db",
        DatabaseHostEnvVar:     "DATABASE_HOST",
        DatabasePortEnvVar:     "DATABASE_PORT",
        DatabaseUserEnvVar:     "DATABASE_USER",
        DatabasePasswordEnvVar: "DATABASE_PASSWORD",
        DatabaseNameEnvVar:     "DATABASE_NAME",
        DatabaseInitScript:     filepath.Join(".", "testAssets", "testdata", "dev-db.sql"),
        EnvironmentVariables: map[string]string{
            "CHECK_ISBN_CLIENT_HOST": "https://api.isbncheck.com",
            "EMAIL_CLIENT_HOST":      "https://api.gmail.com",
        },
    }
}

func NewMainWithTestContainers(ctx context.Context) *TestContainersContext {
    params := NewTestContainersParams()
    // Set the environment variables
    setEnvVars(params.EnvironmentVariables)
    // Start the postgres container
    initPostgresContainer(ctx, params)
    // Create database connection
    db := getDatabaseConnectionTestContainers(params)
    // Mock the third-party API client
    mockClient := &http.Client{}
    httpmock.ActivateNonDefault(mockClient)
    // Build the app
    server, _ := mainHttpServerSetup(params.MainHttpServerAddress, mockClient)
    return &TestContainersContext{
        MainHttpServer: server,
        Database:       db,
        Params:         params,
    }
}

func getDatabaseConnectionTestContainers(params *TestContainersParams) *sql.DB {
    db, err := sql.Open("postgres", fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
        os.Getenv(params.DatabaseHostEnvVar),
        os.Getenv(params.DatabasePortEnvVar),
        os.Getenv(params.DatabaseUserEnvVar),
        os.Getenv(params.DatabasePasswordEnvVar),
        os.Getenv(params.DatabaseNameEnvVar),
    ))
    if err != nil {
        log.Fatalf("failed to connect to database: %s", err)
    }
    return db
}

// initPostgresContainer starts a postgres container and sets the required environment variables.
// Source: https://golang.testcontainers.org/modules/postgres/
func initPostgresContainer(ctx context.Context, params *TestContainersParams) *postgres.PostgresContainer {
    logTimeout := 10
    port := "5432/tcp"
    dbURL := func(host string, port nat.Port) string {
        return fmt.Sprintf("postgres://postgres:postgres@%s:%s/%s?sslmode=disable",
            host, port.Port(), params.DatabaseName)
    }

    postgresContainer, err := postgres.Run(ctx, params.PostgresImage,
        postgres.WithInitScripts(params.DatabaseInitScript),
        postgres.WithDatabase(params.DatabaseName),
        testcontainers.WithWaitStrategy(wait.ForSQL(nat.Port(port), "postgres", dbURL).
            WithStartupTimeout(time.Duration(logTimeout)*time.Second)),
    )
    if err != nil {
        log.Fatalf("failed to start postgresContainer: %s", err)
    }
    postgresHost, _ := postgresContainer.Host(ctx) //nolint:errcheck // non-critical
    // Set database environment variables
    setEnvVars(map[string]string{
        params.DatabaseHostEnvVar:     postgresHost,
        params.DatabasePortEnvVar:     getPostgresPort(ctx, postgresContainer),
        params.DatabaseUserEnvVar:     "postgres",
        params.DatabasePasswordEnvVar: "postgres",
        params.DatabaseNameEnvVar:     params.DatabaseName,
    })

    s, err := postgresContainer.ConnectionString(ctx)
    log.Printf("Postgres container started at: %s", s)
    if err != nil {
        log.Fatalf("error from initPostgresContainer - ConnectionString: %e", err)
    }

    return postgresContainer

}

func getPostgresPort(ctx context.Context, postgresContainer *postgres.PostgresContainer) string {
    port, e := postgresContainer.MappedPort(ctx, "5432/tcp")
    if e != nil {
        log.Fatalf("Failed to get postgresContainer.Ports: %s", e)
    }

    return port.Port()
}

// setEnvVars sets environment variables from a map.
func setEnvVars(envs map[string]string) {
    for key, value := range envs {
        if err := os.Setenv(key, value); err != nil {
            log.Fatalf("Failed to set environment variable %s: %v", key, err)
        }
    }
}

7. Configure cucumber to run the integration test

integration_test.go

package main

import (
    "context"
    "log"
    "net/http"
    "testing"
    "time"

    "github.com/cucumber/godog"
)

func TestFeatures(t *testing.T) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Initialize test containers configuration
    testcontainersConfig := NewMainWithTestContainers(ctx)

    // Channel to notify when the server is ready
    serverReady := make(chan struct{})

    // Start the HTTP server in a separate goroutine
    go func() {
        log.Println("Listening for requests at http://localhost" + testcontainersConfig.Params.MainHttpServerAddress)
        // Notify that the server is ready
        close(serverReady)

        if err := testcontainersConfig.MainHttpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Error starting the server: %v", err)
        }
    }()

    // Wait for the server to start
    <-serverReady

    // Allow a brief moment for the server to initialize
    time.Sleep(500 * time.Millisecond)

    // Run the godog test suite
    suite := godog.TestSuite{
        ScenarioInitializer: func(sc *godog.ScenarioContext) {
            // This address should match the address of the app in the testcontainers_config.go file
            mainHttpServerUrl := "http://localhost" + testcontainersConfig.Params.MainHttpServerAddress
            NewStepsContext(mainHttpServerUrl, testcontainersConfig.Database, sc)
        },
        Options: &godog.Options{
            Format:   "pretty",
            Paths:    []string{"features"}, // Edit this path locally to execute only the feature files you want to test.
            TestingT: t,                    // Testing instance that will run subtests.
        },
    }

    if suite.Run() != 0 {
        t.Fatal("Non-zero status returned, failed to run feature tests")
    }

    // Gracefully shutdown the server after tests are done
    if err := testcontainersConfig.MainHttpServer.Shutdown(ctx); err != nil {
        log.Fatalf("Error shutting down the server: %v", err)
    }
}

8. Execute the test

go test -v
=== RUN   TestFeatures
2024/09/12 16:09:52 github.com/testcontainers/testcontainers-go - Connected to docker: 
  Server Version: 24.0.7
  API Version: 1.43
  Operating System: Ubuntu 23.10
  Total Memory: 5910 MB
  Testcontainers for Go Version: v0.33.0
  Resolved Docker Host: unix:////Users/jose.boretto/.colima/docker.sock
  Resolved Docker Socket Path: /var/run/docker.sock
  Test SessionID: 1c9d3b9a409b2a340378c42d67678f67249a8f748d346a0dd3758b2ce56c3bda
  Test ProcessID: f1928595-5aaa-4f92-9c93-e2f2bad03364
2024/09/12 16:09:52 🐳 Creating container for image testcontainers/ryuk:0.8.1
2024/09/12 16:09:52 ✅ Container created: c0b1cf852231
2024/09/12 16:09:52 🐳 Starting container: c0b1cf852231
2024/09/12 16:09:52 ✅ Container started: c0b1cf852231
2024/09/12 16:09:52 ⏳ Waiting for container id c0b1cf852231 image: testcontainers/ryuk:0.8.1. Waiting for: &{Port:8080/tcp timeout:<nil> PollInterval:100ms skipInternalCheck:false}
2024/09/12 16:09:53 🔔 Container is ready: c0b1cf852231
2024/09/12 16:09:53 🐳 Creating container for image docker.io/postgres:16-alpine
2024/09/12 16:09:53 ✅ Container created: 3980cd9f254e
2024/09/12 16:09:53 🐳 Starting container: 3980cd9f254e
2024/09/12 16:09:53 ✅ Container started: 3980cd9f254e
2024/09/12 16:09:53 ⏳ Waiting for container id 3980cd9f254e image: docker.io/postgres:16-alpine. Waiting for: &{timeout:<nil> deadline:0x14000011148 Strategies:[0x140000aa780]}
2024/09/12 16:09:56 🔔 Container is ready: 3980cd9f254e
2024/09/12 16:09:56 Postgres container started at: postgres://postgres:postgres@localhost:32809/db?
databaseConnectionString:  host=localhost user=postgres password=postgres dbname=db port=32809 sslmode=disable
2024/09/12 16:09:56 Listening for requests at http://localhost:8000
Feature: Create book
=== RUN   TestFeatures/Create_a_new_book_successfully

  Background: Clean database
    Given SQL command                                                                                                                                  # <autogenerated>:1 -> *StepsContext
      """
      DELETE FROM myschema.books;
      """
    And reset mock server                                                                                                                              # <autogenerated>:1 -> *StepsContext

  Scenario: Create a new book successfully                                                                                                             # features/createBook.feature:10
    Given a mock server request with method: "GET" and url: "https://api.isbncheck.com/isbn/0-061-96436-1"                                             # <autogenerated>:1 -> *StepsContext
    And a mock server response with status 200 and body                                                                                                # <autogenerated>:1 -> *StepsContext
      """ json
      {
         "id": "0-061-96436-1"
      }
      """
    And a mock server request with method: "POST" and url: "https://api.gmail.com/send-email" and body                                                 # <autogenerated>:1 -> *StepsContext
      """ json
      {
        "email" : "helloworld@gmail.com",
        "book" : {
          "isbn" : "0-061-96436-1",
          "title" : "The Art of Computer Programming"
        }
      }
      """
    And a mock server response with status 200 and body                                                                                                # <autogenerated>:1 -> *StepsContext
      """ json
      {
         "status": "OK"
      }
      """

2024/09/12 16:09:57 /Users/jose.boretto/Documents/jose/golang-testcontainers-gherkin-setup/internal/infrastructure/persistance/book/create_book_repository.go:40 record not found
[3.278ms] [rows:0] SELECT * FROM "myschema"."books" WHERE isbn = '0-061-96436-1' AND "books"."deleted_at" IS NULL ORDER BY "books"."id" LIMIT 1
    When API "POST" request is sent to "/api/v1/createBook" with payload                                                                               # <autogenerated>:1 -> *StepsContext
      """ json
      {
        "isbn": "0-061-96436-1",
        "title": "The Art of Computer Programming"
      }
      """
    Then API response status code is 200 and payload is                                                                                                # <autogenerated>:1 -> *StepsContext
      """ json
      {
          "isbn": "0-061-96436-1",
          "title": "The Art of Computer Programming"
      }
      """
    And SQL query "SELECT * FROM myschema.books WHERE isbn = '0-061-96436-1'" result without the fields "created_at,deleted_at,updated_at" is equal to # <autogenerated>:1 -> *StepsContext
      """ json
      [
         {
            "id":1,
            "isbn":"0-061-96436-1",
            "title":"The Art of Computer Programming"
         }
      ]
      """

1 scenarios (1 passed)
9 steps (9 passed)
30.273417ms
--- PASS: TestFeatures (4.66s)
    --- PASS: TestFeatures/Create_a_new_book_successfully (0.03s)
PASS
ok      github.com/joseboretto/golang-testcontainers-gherkin-setup/cmd  5.087s

Conclusion

Combining Cucumber, Testcontainers, and HTTPMock gives you a powerful suite of tools for writing integration tests in
Go. With this setup, you can ensure your application is well-tested, reliable, and ready for production deployment.

By leveraging the strengths of each tool, you can achieve more comprehensive test coverage and ensure that all parts of
your system work together seamlessly.

Feel free to experiment with these tools and extend the example to suit your own application’s needs!

Here you can find the repository with the
code: https://github.com/joseboretto/golang-testcontainers-gherkin-setup