En la lección anterior conocimos Testcontainers: una herramienta que permite levantar bases de datos reales y otros servicios (incluyendo Kafka y RabbitMQ) en el entorno de pruebas. Esto será clave en la práctica de hoy.
¿Por qué probar procesos asíncronos es un reto?
Los procesos asíncronos le dan su toque a las pruebas. A diferencia de las "peticiones síncronas normales", donde el resultado se devuelve inmediatamente (o no se devuelve si todo se rompe), aquí tratamos con:
- Colas de mensajes: Transferencia de información a través de brokers de datos (Kafka, RabbitMQ).
- Retrasos: A veces los mensajes se procesan con latencia.
- Verificación de varios componentes: Producers, consumers y el propio broker (que también tiene sus manías).
Tareas principales al probar procesos asíncronos:
- Asegurarse de que el mensaje fue enviado correctamente por el producer.
- Comprobar que el consumer procesó el mensaje correctamente.
- Verificar que los mensajes no se pierden y cumplen las garantías de entrega necesarias.
Herramientas para probar Kafka y RabbitMQ
Para testear procesos asíncronos usaremos:
- JUnit 5: Para escribir tests.
- Testcontainers: Para levantar instancias aisladas de Kafka y RabbitMQ.
- Spring Kafka Test: Librería para testear Kafka en Spring Boot.
- Spring AMQP Test: Librería para testear RabbitMQ.
- Awaitility: Para esperar cómodamente resultados de operaciones asíncronas.
¿Cómo probar Kafka?
1. Preparar el entorno con Testcontainers
Testcontainers es perfecto para ejecutar Kafka en el entorno de pruebas. Empecemos con la configuración.
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.utility.DockerImageName;
import static org.junit.jupiter.api.Assertions.assertNotNull;
public class KafkaTest {
@Test
void testKafkaContainer() {
// Arrancamos Kafka con Testcontainers
KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));
kafkaContainer.start();
// Comprobamos que Kafka está en ejecución
assertNotNull(kafkaContainer.getBootstrapServers(), "Kafka no está en ejecución");
kafkaContainer.stop();
}
}
¿Qué hicimos?
- Arrancamos Kafka dentro de un contenedor Docker.
- Comprobamos que Kafka arrancó y que podemos obtener la dirección de su servidor.
2. Testear el producer
Ahora creamos un test para verificar que los mensajes se envían correctamente a Kafka.
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.beans.factory.annotation.Autowired;
import static org.assertj.core.api.Assertions.assertThat;
@EmbeddedKafka(partitions = 1, topics = {"test-topic"})
public class KafkaProducerTest {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Test
void testSendMessage() throws Exception {
// Enviamos el mensaje a Kafka
ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key", "message");
RecordMetadata metadata = kafkaTemplate.send(record).get().getRecordMetadata();
// Comprobamos que el mensaje se envió correctamente
assertThat(metadata.topic()).isEqualTo("test-topic");
}
}
¿Qué hicimos?
- Usamos un broker Kafka embebido (
@EmbeddedKafka) para las pruebas, así no hay que levantar la infraestructura de test manualmente. - Enviamos un mensaje usando
KafkaTemplatey verificamos que se escribió correctamente en el topic.
3. Testear el consumer
Los consumers son nuestros "currantes" que procesan mensajes. Vamos a comprobar que funcionan correctamente.
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@EmbeddedKafka(partitions = 1, topics = {"test-topic"})
public class KafkaConsumerTest {
@Test
void testConsumeMessage() {
// Configuramos el consumer de prueba
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", kafkaBroker);
DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<String, String> consumer = consumerFactory.createConsumer();
// Nos suscribimos al topic y verificamos el mensaje recibido
consumer.subscribe(Collections.singletonList("test-topic"));
ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, "test-topic");
assertThat(record.value()).isEqualTo("message");
}
}
¿Cómo probar RabbitMQ?
1. Preparar el entorno
Levantemos RabbitMQ usando Testcontainers.
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class RabbitMQTest {
@Test
void testRabbitMQContainer() {
RabbitMQContainer rabbitMQContainer = new RabbitMQContainer("rabbitmq:3-management");
rabbitMQContainer.start();
assertTrue(rabbitMQContainer.isRunning());
rabbitMQContainer.stop();
}
}
2. Testear el producer
Verificamos que el mensaje se envía a la queue de RabbitMQ.
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class RabbitProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void testSendMessage() {
// Enviamos el mensaje
String queue = "test-queue";
String message = "Hello RabbitMQ";
rabbitTemplate.convertAndSend(queue, message);
// Verificamos que el mensaje fue enviado
String receivedMessage = (String) rabbitTemplate.receiveAndConvert(queue);
assertEquals(message, receivedMessage);
}
}
3. Testear el consumer
Escribamos un test para comprobar el consumer.
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class RabbitConsumerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private MessageListenerContainer container;
@Test
void testConsumeMessage() {
// Enviamos el mensaje
String queue = "test-queue";
String message = "Hello RabbitMQ";
rabbitTemplate.convertAndSend(queue, message);
// Verificamos que el mensaje fue procesado
container.start();
String receivedMessage = (String) rabbitTemplate.receiveAndConvert(queue);
assertTrue(receivedMessage.equals(message));
container.stop();
}
}
¿Qué es importante recordar?
- La asincronía es complicada: ten en cuenta las latencias en el procesamiento de mensajes. Usa librerías como Awaitility para esperar.
- No olvides aislar los tests: Testcontainers o @EmbeddedKafka/@RabbitMQ te ayudarán con eso.
- Kafka y RabbitMQ son colas de mensajes con diferencias, por lo que los enfoques de testing pueden variar un poco.
Ahora estás equipado con lo necesario para probar procesos asíncronos.
GO TO FULL VERSION