Understanding End-to-End Testing

E2E testing for Alert Producer and Alert Consumer validates the complete process of creating and delivering alerts. It ensures that alerts are accurately transmitted from the producer to Kafka, and subsequently received and processed by the consumer, confirming that the system operates as intended in real-world scenarios.

To facilitate E2E testing, we will employ Testcontainers to orchestrate Docker containers for Zookeeper, Kafka, Schema Registry, Alert Producer, and Alert Consumer.

Building Docker Images

Creating the Alert Producer Docker Image

Open a terminal in the root directory of the alert-producer and execute the following command:

./mvnw clean spring-boot:build-image -DskipTests

This will create the Docker image alert-producer:0.0.1-SNAPSHOT.

Creating the Alert Consumer Docker Image

Next, navigate to the alert-consumer's root directory in a terminal and run:

./mvnw clean spring-boot:build-image -DskipTests

This command will generate the alert-consumer:0.0.1-SNAPSHOT Docker image.

Setting Up the E2E Testing Spring Boot Application

We will establish a Spring Boot application using Spring Initializr. Name the application alert-e2e-testing, including the required dependencies: Spring Web and Testcontainers. We will use Spring Boot version 3.2.4 and Java 17. Use the provided link for all previous setup details. Click the GENERATE button to download a zip file, then extract it into the alert-app folder and open the alert-e2e-testing project in your IDE.

Creating the AbstractTestcontainers Class

In the src/test/java directory, create the AbstractTestcontainers class under the com.example.alerte2etesting package. The class should contain the following code:

package com.example.alerte2etesting;

import org.springframework.test.context.DynamicPropertyRegistry;

import org.springframework.test.context.DynamicPropertySource;

import org.testcontainers.containers.GenericContainer;

import org.testcontainers.containers.KafkaContainer;

import org.testcontainers.containers.Network;

import org.testcontainers.containers.wait.strategy.Wait;

import org.testcontainers.junit.jupiter.Testcontainers;

import org.testcontainers.utility.DockerImageName;

import java.time.Duration;

@Testcontainers

public abstract class AbstractTestcontainers {

private static final KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.3"));

private static final GenericContainer schemaRegistryContainer = new GenericContainer<>("confluentinc/cp-schema-registry:7.5.3");

protected static final GenericContainer alertProducerContainer = new GenericContainer<>("alert-producer:0.0.1-SNAPSHOT");

protected static final GenericContainer alertConsumerContainer = new GenericContainer<>("alert-consumer:0.0.1-SNAPSHOT");

protected static final int ALERT_PRODUCER_EXPOSED_PORT = 8080;

@DynamicPropertySource

private static void dynamicProperties(DynamicPropertyRegistry registry) {

Network network = Network.SHARED;

kafkaContainer.withNetwork(network)

.withNetworkAliases("kafka")

.start();

schemaRegistryContainer.withNetwork(network)

.withNetworkAliases("schema-registry")

.withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092")

.withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")

.withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8081")

.waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(1)))

.start();

alertProducerContainer.withNetwork(network)

.withNetworkAliases("alert-producer")

.withEnv("KAFKA_HOST", "kafka")

.withEnv("KAFKA_PORT", "9092")

.withEnv("SCHEMA_REGISTRY_HOST", "schema-registry")

.withExposedPorts(ALERT_PRODUCER_EXPOSED_PORT)

.waitingFor(Wait.forLogMessage(".*Tomcat started on port.*", 1))

.start();

alertConsumerContainer.withNetwork(network)

.withNetworkAliases("alert-consumer")

.withEnv("KAFKA_HOST", "kafka")

.withEnv("KAFKA_PORT", "9092")

.withEnv("SCHEMA_REGISTRY_HOST", "schema-registry")

.waitingFor(Wait.forLogMessage(".*Tomcat started on port.*", 1))

.start();

}

}

The @Testcontainers annotation indicates that this class will interact with Testcontainers. It establishes four containers: Kafka, Schema Registry, Alert Producer, and Alert Consumer, which are configured and initiated in the dynamicProperties() method.

Updating the AlertE2eTestingApplicationTests Class

Next, we will revise the AlertE2eTestingApplicationTests class located in the src/test/java directory with the following content:

package com.example.alerte2etesting;

import org.junit.jupiter.api.Test;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;

import org.springframework.boot.test.web.client.TestRestTemplate;

import org.springframework.http.HttpStatus;

import org.springframework.http.ResponseEntity;

import org.testcontainers.containers.output.OutputFrame;

import org.testcontainers.containers.output.WaitingConsumer;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;

import static org.assertj.core.api.Assertions.assertThat;

import static org.assertj.core.api.Assertions.fail;

@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)

class AlertE2eTestingApplicationTests extends AbstractTestcontainers {

@Autowired

private TestRestTemplate testRestTemplate;

@Test

void testEmitAlert() {

WaitingConsumer waitingConsumer = new WaitingConsumer();

alertConsumerContainer.followOutput(waitingConsumer, OutputFrame.OutputType.STDOUT);

String alertProducerApiUrl = "http://localhost:%s/api/alerts".formatted(alertProducerContainer.getMappedPort(ALERT_PRODUCER_EXPOSED_PORT));

AlertDto alertDto = new AlertDto("Alert Test 123456 ABCDEF", "YELLOW");

ResponseEntity responseEntity = testRestTemplate.postForEntity(alertProducerApiUrl, alertDto, Void.class);

assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.CREATED);

try {

waitingConsumer.waitUntil(frame ->

frame.getUtf8String().contains("Received Alert! "Alert Test 123456 ABCDEF" with color 'YELLOW'"), 5, TimeUnit.SECONDS);

} catch (TimeoutException e) {

fail("The expected message was not received");

}

}

private record AlertDto(String description, String color) {

}

}

Here, the @SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT) annotation instructs Spring Boot to initiate the application on a random port for testing. This class also inherits the shared testing setup from the AbstractTestcontainers class. The testEmitAlert() method verifies the end-to-end workflow of sending an alert from the producer to the consumer, employing TestRestTemplate for HTTP requests to the application.

Executing the E2E Test

To run the tests, open a terminal, navigate to the root folder of alert-e2e-testing, and execute:

./mvnw clean test

Testcontainers will launch the Zookeeper, Kafka, Schema Registry, Alert Producer, and Alert Consumer Docker containers before executing the E2E tests. All tests should pass upon completion.

Additional Resources

If you found this guide helpful and wish to support my work, consider taking the following actions:

👏 Engage by clapping, highlighting, or responding to my posts; I'm here to answer any questions you have.

🌐 Share this guide on social media.

🔔 Follow me on Medium, LinkedIn, Twitter, and GitHub.

✉️ Subscribe to my newsletter to stay updated on my latest articles.

Spring Boot | Kafka Schema Registry & Avro with Practical Example

This video demonstrates how to integrate Spring Boot applications with Kafka, showcasing the use of Schema Registry and Avro for message serialization.

Spring for Apache Kafka® 101: Confluent Cloud Schema Registry and Spring Boot (Hands On)

This tutorial provides a hands-on approach to using Confluent Cloud Schema Registry with Spring Boot applications, focusing on practical implementation.