Как макетировать конечную точку потребителя Kafka на верблюжьем маршруте? - PullRequest
Новая
       30

Как макетировать конечную точку потребителя Kafka на верблюжьем маршруте?

0 голосов
/ 03 апреля 2020

У меня есть конечная точка Camel, которая по сути является потребителем Kafka, читающим из топи c и отправляющим информацию в базу данных. Он работает нормально, однако я изо всех сил пытаюсь выполнить его модульное тестирование, так как я не смог издеваться над конечной точкой Кафки. Может ли кто-нибудь помочь мне издеваться над потребителем Кафки на верблюжьем маршруте? 

@Override
public void configure() {    
    from(kafka:eph?brokers=localhost:9092...).routeId("KafkaConsumer")
        .to(direct:updateDatabase)
}

Ответы [ 2 ]

0 голосов
/ 08 апреля 2020

Просто экстернализуйте URI конечной точки в свойстве (например, с помощью свойства Spring) 

from(consumerEndpoint).routeId("KafkaConsumer")

Затем в вашей производственной конфигурации вы используете реальную конечную точку 

consumerEndpoint=kafka:eph?brokers=localhost:9092...

Тогда как в В вашей тестовой конфигурации вы используете прямую конечную точку 

consumerEndpoint=direct:consumer

. Эту настройку легко запустить из теста маршрута Camel 

producer.sendBody("direct:consumer", myMessageBody);
.
0 голосов
/ 03 апреля 2020

Для модульного тестирования вашего маршрута вы можете сделать это с помощью стандартного теста на загрузку с верблюжьей пружиной. Во время теста производитель Kafka (с точки зрения Camel) может быть заменен прямым компонентом, и ложные сообщения могут доставляться туда. Чтобы проверить, правильно ли ваши маршруты обрабатывают эти сообщения, можно использовать Макетные конечные точки . 

//Route definition
@Component
public class KafkaRoute extends RouteBuilder {

    public static final String KAFKA_ROUTE_NAME = "kafka-route";

    @Override
    public void configure() throws Exception {
        from("kafka:eph?brokers=localhost:9092").routeId(KAFKA_ROUTE_NAME)
                .log(LoggingLevel.INFO, "Message: ${body}  received on the topic: ${headers[kafka.TOPIC]} ")
                .to("direct:updateDatabase");

        from("direct:updateDatabase").log(LoggingLevel.INFO, "DB Updated.");

    }

}


import java.util.HashMap;
import java.util.Map;

import org.apache.camel.CamelContext;
import org.apache.camel.EndpointInject;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.AdviceWithRouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.spring.CamelSpringBootRunner;
import org.apache.camel.test.spring.MockEndpoints;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.annotation.DirtiesContext;

@RunWith(CamelSpringBootRunner.class)
@SpringBootTest
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@MockEndpoints("direct:*")
public class KafkaRouteTest {

    @Autowired
    CamelContext camelContext;

    @Produce
    ProducerTemplate mockKafkaProducer;

    @EndpointInject("mock:direct:updateDatabase")
    MockEndpoint finalSink;

    @Test
    public void testKafkaRoute() throws Exception {

        //Here we swap the FROM component in the KafkaRoute.KAFKA_ROUTE_NAME with a direct component, direct:kafka-from
        AdviceWithRouteBuilder.adviceWith(camelContext, KafkaRoute.KAFKA_ROUTE_NAME, routeBuilder -> {
            routeBuilder.replaceFromWith("direct:kafka-from");
        });

        Map<String, Object> headers = new HashMap<>();
        headers.put(KafkaConstants.TOPIC, "testTopic");

        //Send mock message to the route
        mockKafkaProducer.sendBodyAndHeaders("direct:kafka-from", "test-body", headers);


        //Assertions. You may do additional assertions with the likes of Mockito
        finalSink.expectedBodiesReceived("test-body");
        finalSink.expectedHeaderReceived(KafkaConstants.TOPIC, "testTopic");
        finalSink.assertIsSatisfied();

    }

}

Компонент Camel Kafka уже протестирован на модуле , там нет смысла копировать все эти тесты в вашей кодовой базе. Однако, если вы действительно хотите провести тестирование на реальном экземпляре Kafka, вы можете использовать тестовые контейнеры . Здесь - это полноценный пример из самого репозитория Camel с использованием тестовых контейнеров.

...