You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2023/01/25 14:14:04 UTC
[camel] branch main updated (ade9e616a1a -> e7825a48c9f)
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
from ade9e616a1a Update Jenkinsfile.s390x
new 8467d5839f3 CAMEL-18957: initial introduction of a JUnit 5 extension for the Camel Context
new 6537f0fd563 CAMEL-18957: convert camel-amqp to the new camel-test-infra-core
new e26777ab82e CAMEL-18957: convert camel-kafka to the new camel-test-infra-core
new e7825a48c9f (chores) camel-test-infra-kafka: improve logging when starting up the service fails
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
components/camel-amqp/pom.xml | 8 +
.../apache/camel/component/amqp/AMQPRouteTest.java | 45 ++--
.../component/amqp/AMQPRouteTraceFrameTest.java | 53 ++--
.../camel/component/amqp/AMQPTestSupport.java | 19 +-
.../component/amqp/AMQPToDSendDynamicTest.java | 38 ++-
.../apache/camel/component/amqp/AMQPToDTest.java | 38 ++-
.../amqp/artemis/AMQPEmbeddedBrokerTest.java | 51 ++--
components/camel-kafka/pom.xml | 12 +
.../kafka/KafkaIdempotentRepository.java | 2 +
.../camel/component/kafka/KafkaAutowireTest.java | 14 +-
.../camel/component/kafka/KafkaComponentTest.java | 18 +-
.../integration/BaseEmbeddedKafkaTestSupport.java | 55 ++--
.../KafkaConsumerAsyncManualCommitIT.java | 39 ++-
.../kafka/integration/KafkaConsumerAuthIT.java | 49 +++-
.../integration/KafkaConsumerAuthInvalidIT.java | 58 +++--
.../KafkaConsumerAuthInvalidWithReconnectIT.java | 63 ++---
.../integration/KafkaConsumerAuthManualTest.java | 30 ++-
...KafkaConsumerAutoInstResumeRouteStrategyIT.java | 22 +-
.../integration/KafkaConsumerBatchSizeIT.java | 20 +-
.../kafka/integration/KafkaConsumerFullIT.java | 58 +++--
.../integration/KafkaConsumerHealthCheckIT.java | 227 -----------------
.../integration/KafkaConsumerIdempotentIT.java | 36 +--
...kaConsumerIdempotentWithCustomSerializerIT.java | 35 +--
.../KafkaConsumerIdempotentWithProcessorIT.java | 33 +--
.../KafkaConsumerLastRecordHeaderIT.java | 6 +-
.../integration/KafkaConsumerRebalanceIT.java | 19 +-
.../kafka/integration/KafkaConsumerStopIT.java | 30 +--
.../integration/KafkaConsumerTopicIsPatternIT.java | 18 +-
.../kafka/integration/KafkaProducerFullIT.java | 212 +++++++---------
.../component/kafka/integration/KafkaToDIT.java | 7 +-
.../kafka/integration/KafkaTransactionIT.java | 82 +++---
.../{ => commit}/BaseManualCommitTestSupport.java | 13 +-
.../commit/KafkaConsumerAsyncCommitIT.java | 28 +-
.../KafkaConsumerAsyncWithOffsetRepoCommitIT.java | 28 +-
.../commit/KafkaConsumerNoopCommitIT.java | 30 ++-
.../commit/KafkaConsumerSyncCommitIT.java | 17 +-
.../KafkaConsumerSyncWithOffsetRepoCommitIT.java | 31 ++-
.../KafkaAdminUtil.java} | 70 ++---
.../KafkaTestUtil.java} | 34 +--
.../kafka/integration/common/TestProducerUtil.java | 48 ++++
.../health/KafkaConsumerBadPortHealthCheckIT.java | 110 ++------
...fkaConsumerBadPortSupervisingHealthCheckIT.java | 140 ++++------
.../health/KafkaConsumerHealthCheckIT.java | 188 ++++++++++++++
.../KafkaConsumerUnresolvableHealthCheckIT.java | 143 +++--------
.../health/KafkaHealthCheckTestSupport.java | 103 ++++++++
.../KafkaPausableConsumerCircuitBreakerIT.java | 75 +++---
.../integration/pause/KafkaPausableConsumerIT.java | 107 ++++----
.../kafka/KafkaIdempotentRepositoryEagerIT.java | 39 ++-
.../kafka/KafkaIdempotentRepositoryNonEagerIT.java | 63 +++--
.../KafkaIdempotentRepositoryPersistenceIT.java | 34 ++-
.../idempotent/kafka/SimpleIdempotentTest.java | 47 ++++
.../user-manual/modules/ROOT/pages/test-infra.adoc | 157 ++++++++++--
docs/user-manual/modules/ROOT/pages/testing.adoc | 4 +-
test-infra/camel-test-infra-core/pom.xml | 80 ++++++
.../test/infra/core/CamelContextExtension.java | 68 +++++
.../camel/test/infra/core/ContextFixture.java | 29 ++-
.../test/infra/core/ContextLifeCycleManager.java | 51 ++++
.../infra/core/DefaultCamelContextExtension.java | 282 +++++++++++++++++++++
.../infra/core/DefaultContextLifeCycleManager.java | 68 +++++
.../apache/camel/test/infra/core/RouteFixture.java | 29 ++-
.../test/infra/core/api/ConfigurableContext.java | 27 +-
.../test/infra/core/api/ConfigurableRoute.java | 26 +-
.../test/infra/kafka/services/KafkaService.java | 14 +-
test-infra/pom.xml | 1 +
64 files changed, 2149 insertions(+), 1432 deletions(-)
delete mode 100644 components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
rename components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/{ => commit}/BaseManualCommitTestSupport.java (90%)
rename components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/{BaseEmbeddedKafkaAuthTestSupport.java => common/KafkaAdminUtil.java} (61%)
rename components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/{AbstractKafkaTestSupport.java => common/KafkaTestUtil.java} (72%)
create mode 100644 components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/TestProducerUtil.java
create mode 100644 components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerHealthCheckIT.java
create mode 100644 components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaHealthCheckTestSupport.java
create mode 100644 components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/SimpleIdempotentTest.java
create mode 100644 test-infra/camel-test-infra-core/pom.xml
create mode 100644 test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/CamelContextExtension.java
copy components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java => test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextFixture.java (55%)
create mode 100644 test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextLifeCycleManager.java
create mode 100644 test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultCamelContextExtension.java
create mode 100644 test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultContextLifeCycleManager.java
copy components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java => test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/RouteFixture.java (55%)
copy components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java => test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableContext.java (55%)
copy components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java => test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableRoute.java (55%)
[camel] 01/04: CAMEL-18957: initial introduction of a JUnit 5 extension for the Camel Context
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8467d5839f37452ae54ad9ed07d525e3c23182d7
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Thu Jan 19 11:53:52 2023 +0100
CAMEL-18957: initial introduction of a JUnit 5 extension for the Camel Context
---
.../user-manual/modules/ROOT/pages/test-infra.adoc | 157 ++++++++++--
docs/user-manual/modules/ROOT/pages/testing.adoc | 4 +-
test-infra/camel-test-infra-core/pom.xml | 80 ++++++
.../test/infra/core/CamelContextExtension.java | 68 +++++
.../camel/test/infra/core/ContextFixture.java | 35 +++
.../test/infra/core/ContextLifeCycleManager.java | 51 ++++
.../infra/core/DefaultCamelContextExtension.java | 282 +++++++++++++++++++++
.../infra/core/DefaultContextLifeCycleManager.java | 68 +++++
.../apache/camel/test/infra/core/RouteFixture.java | 35 +++
.../test/infra/core/api/ConfigurableContext.java | 35 +++
.../test/infra/core/api/ConfigurableRoute.java | 34 +++
test-infra/pom.xml | 1 +
12 files changed, 823 insertions(+), 27 deletions(-)
diff --git a/docs/user-manual/modules/ROOT/pages/test-infra.adoc b/docs/user-manual/modules/ROOT/pages/test-infra.adoc
index 8c4135a36ff..d3300b1370a 100644
--- a/docs/user-manual/modules/ROOT/pages/test-infra.adoc
+++ b/docs/user-manual/modules/ROOT/pages/test-infra.adoc
@@ -1,5 +1,71 @@
= Test Infrastructure
+The components in the Camel Test Infra provide utilities to simplify testing with Camel and other systems may interact with it. They work as JUnit 5 extensions.
+
+== Working with the Camel Context in Tests
+
+When testing Camel or a Camel-based integration, you almost certainly need to use the `CamelContext` to configure the registry, add routes and execute other operations. The test infra comes with a module that provides a JUnit 5 extension that allows you to inject a Camel context into your tests.
+
+Adding it to your test code is as simple as adding the following lines of code to your test class:
+
+[source,java]
+----
+@RegisterExtension
+protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+----
+
+Then, via the extension, you can access the context (ie.: ```contextExtension.getContext()`) to manipulate it as need in the tests.
+
+The extension comes with a few utilities to simplify configuring the context, and adding routes at the appropriate time.
+
+=== Configuring the Camel Context
+
+To create a method that configures the context, you can declare a method receiving an instance of `CamelContext` and annotate it with `@ContextFixture`.
+
+[source,java]
+----
+@ContextFixture
+public void configureContext(CamelContext context) {
+ // context configuration code here
+}
+----
+
+Additionally, you can simplify the class hierarchy, and ensure consistency you may also implement the `ConfigurableContext` interface.
+
+=== Configuring the Routes
+
+You can configure the routes using a similar process as the one described for configuring the Camel context. You can create a method that receives an instance of `CamelContext` and annotate it with `@RouteFixture`.
+
+[source,java]
+----
+@RouteFixture
+public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(fromUri).to(destUri);
+ }
+ });
+}
+----
+
+=== Use the Camel Context Extension
+
+To start using the Camel Context extension on your code, add the following dependency:
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-core</artifactId>
+ <version>${camel.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+</dependency>
+----
+
+For simplicity and consistency, you may also declare the route as implementing the `ConfigurableRoute`.
+
== Simulating the Test Infrastructure
One of the first steps when implementing a new test, is to identify how to simulate infrastructure required for it to
@@ -10,7 +76,13 @@ and uses container images to simulate the environments. Additionally, it may als
environments as well as, when available, embeddable components. This varies by each component, and it is recommended to
check the code for additional details.
-== Writing A New Test Infrastructure Module
+=== Writing A New Test Infrastructure Module
+
+
+[NOTE]
+====
+This section is aimed at Camel maintainers that need to write new test infra components. End users can skip this section.
+====
The test code abstracts the provisioning of test environments behind service classes (i.e.: JMSService, JDBCService,
etc). The purpose of the service class is to abstract the both the type service (i.e.: Kafka, Strimzi, etc) and
@@ -31,7 +103,7 @@ whenever possible.
It is also possible to use embeddable components when required, although this usually lead to more code and higher
maintenance.
-=== Recommended Structure for Test Infrastructure Modules
+==== Recommended Structure for Test Infrastructure Modules
The service should provide an interface, named after the infrastructure being implemented, and this interface should
extend the https://github.com/apache/camel/blob/main/test-infra/camel-test-infra-common/src/test/java/org/apache/camel/test/infra/common/services/TestService.java[TestService]
@@ -41,9 +113,12 @@ and https://junit.org/junit5/docs/5.1.1/api/org/junit/jupiter/api/extension/Afte
should be the preferred extensions whenever possible because they allow the instance of the infrastructure to be static
throughout the test execution.
-*Note*: bear in mind that, according to the https://junit.org/junit5/docs/5.1.1/api/org/junit/jupiter/api/extension/RegisterExtension.html[JUnit 5 extension]
+[NOTE]
+====
+Bear in mind that, according to the https://junit.org/junit5/docs/5.1.1/api/org/junit/jupiter/api/extension/RegisterExtension.html[JUnit 5 extension]
model, the time of initialization of the service may differ depending on whether the service instance is declared as
static or not in the test class. As such, the code should make no assumptions as to its time of initialization.
+====
Ideally, there should be two concrete implementations of the services: one of the remote service (if applicable) and
another for the container service:
@@ -66,7 +141,7 @@ as `<name>.<parameter>`. More complex services may use the builder available thr
the service accordingly.
-=== Registering Properties
+==== Registering Properties
All services should register the properties, via `System.setProperty` that allow access to the services. This is required
in order to resolve those properties when running tests using the Spring framework. This registration allows the properties
@@ -74,11 +149,12 @@ to be resolved in Spring's XML files.
This registration is done in the `registerProperties` methods during the service initialization.
-=== Registering Properties Example:
+==== Registering Properties Example:
-Registering the properties in the concrete service implementation:
+Registering the properties in the concrete service implementation:
-```
+[source,java]
+----
public void registerProperties() {
// MyServiceProperties.MY_SERVICE_HOST is a string with value "my.service.host"
System.setProperty(MyServiceProperties.MY_SERVICE_HOST, container.getHost());
@@ -97,25 +173,26 @@ Registering the properties in the concrete service implementation:
registerProperties();
LOG.info("MyService instance running at {}", getServiceAddress());
}
-```
+----
Then, when referring these properties in Camel routes or Spring XML properties, you may use `{{my.service.host}}`,
`{{my.service.port}}` and `{{my.service.address}}`.
-=== Packaging Recommendations
+==== Packaging Recommendations
This is test infrastructure code, therefore it should be package as test type artifacts. The
https://github.com/apache/camel/blob/main/test-infra/camel-test-infra-parent[parent pom] should provide all the necessary bits for packaging the test infrastructure.
-== Using The New Test Infrastructure
+=== Using The New Test Infrastructure
-Using the test infra in a new component test is rather straightforward similar to using any other reusable component.
+Using the test infra in a new component test is rather straightforward, similar to using any other reusable component.
You start by declaring the test infra dependencies in your pom file.
This should be similar to:
-```xml
+[source,xml]
+----
<!-- test infra -->
<dependency>
<groupId>org.apache.camel</groupId>
@@ -124,22 +201,27 @@ This should be similar to:
<type>test-jar</type>
<scope>test</scope>
</dependency>
-```
+----
-*Note*: on the dependencies above, the dependency version is set to `${project.version}`. This should be adjusted to the
+[NOTE]
+====
+On the dependencies above, the dependency version is set to `${project.version}`. This should be adjusted to the
Camel version when used outside the Camel Core project.
+====
On the test class, add a member variable for the service and annotate it with the https://junit.org/junit5/docs/5.1.1/api/org/junit/jupiter/api/extension/RegisterExtension.html[@RegisterExtension],
-in order to let JUnit 5 manage its lifecycle.
+in order to let JUnit 5 manage its lifecycle.
-```
+[source,java]
+----
@RegisterExtension
static MyService service = MyServiceServiceFactory.createService();
-```
+----
-More complex test services can be created using something similar to:
+More complex test services can be created using something similar to:
-```
+[source,java]
+----
@RegisterExtension
static MyService service = MyServiceServiceFactory
.builder()
@@ -147,19 +229,21 @@ static MyService service = MyServiceServiceFactory
.addLocalMapping(MyTestClass::staticMethodReturningAService) // sets the handler for -Dmy-service.instance.type=local-myservice-local-container
.addMapping("local-alternative-service", MyTestClass::anotherMethodReturningAService) // sets the handler for -Dmy-service.instance.type=local-alternative-service
.createService();
-```
+----
You can use the methods as well as the registered properties to access the test infrastructure services available.
-When using these properties in Spring XML files, you may use those properties.
+When using these properties in Spring XML files, you may use those properties.
-```
+[source,xml]
+----
<someSpringXmlElement httpHost="{{my.service.host}}" port="{{my.service.port}}" />
-```
+----
It's also possible to use these properties in the test code itself. For example, when setting up the test url for the
Camel component:
-```
+[source,java]
+----
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
public void configure() {
@@ -168,11 +252,34 @@ Camel component:
}
};
}
-```
+----
+
+==== Execution Ordering
+
+When combining the different modules of the test infra, you may need to ensure that they execute in the proper order. You can do so by using JUnit's `@Order` annotation.
+
+For instance:
+
+[source,java]
+----
+ @Order(1)
+ @RegisterExtension
+ protected static KafkaService service = KafkaServiceFactory.createSingletonService();
+
+ @Order(2)
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+----
== Converting Camel TestContainers Code To The New Test Infrastructure
+
+[NOTE]
+====
+This section is aimed at Camel maintainers that need to write new test infra components. End users can skip this section.
+====
+
Using the camel-nats as an example, we can compare how the base test class for nats changed between https://github.com/apache/camel/blob/camel-3.6.0/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java[3.6.x]
and https://github.com/apache/camel/blob/camel-3.7.0/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java[3.7.x].
diff --git a/docs/user-manual/modules/ROOT/pages/testing.adoc b/docs/user-manual/modules/ROOT/pages/testing.adoc
index c126571cad9..10cec2e6b77 100644
--- a/docs/user-manual/modules/ROOT/pages/testing.adoc
+++ b/docs/user-manual/modules/ROOT/pages/testing.adoc
@@ -20,14 +20,14 @@ The following modules are supported:
|=======================================================================
|Component |Description
-|xref:components:others:test-junit5.adoc[camel-test-junit5] |*JUnit 5*: Is a standalone Java
+|xref:components:others:test-junit5.adoc[camel-test-junit5] |*JUnit 5*: Is an older standalone Java
library letting you easily create Camel test cases using a single Java
class for all your configuration and routing without.
|xref:components:others:test-main-junit5.adoc[camel-test-main-junit5] | *JUnit 5*: Used for testing Camel in Camel Main mode
|xref:components:others:test-spring-junit5.adoc[camel-test-spring-junit5] | *JUnit 5*: Used for testing Camel with Spring / Spring Boot
-|xref:test-infra.adoc[camel-test-infra] | *Camel Test Infra*: Camel Test Infra is a set of modules that leverage Test Containers and abstract the provisioning and execution of test infrastructure. It is the successor of the camel-testcontainers components.
+|xref:test-infra.adoc[camel-test-infra] | *Camel Test Infra*: Camel Test Infra is a set of modules that leverage modern JUnit 5 features to abstract the provisioning and execution of test infrastructure. Among other things, it provides abstraction of the infrastructure (based on Test Containers - being the de-facto successor of the camel-testcontainers components) as well as JUnit 5 extensions for the Camel Context itself.
|=======================================================================
diff --git a/test-infra/camel-test-infra-core/pom.xml b/test-infra/camel-test-infra-core/pom.xml
new file mode 100644
index 00000000000..4ffa7507ec2
--- /dev/null
+++ b/test-infra/camel-test-infra-core/pom.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>test-infra</artifactId>
+ <version>4.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-test-infra-core</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Test Infra :: Core</name>
+ <description>Core testing infrastructure for Camel components</description>
+
+ <properties>
+ <firstVersion>4.0.0</firstVersion>
+ <label>core</label>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- JUnit 5 dependencies -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/CamelContextExtension.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/CamelContextExtension.java
new file mode 100644
index 00000000000..1ddceac2b98
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/CamelContextExtension.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.NoSuchEndpointException;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+
+/**
+ * A JUnit 5 extension that allows you to include a {@link CamelContext} in your test code.
+ */
+public interface CamelContextExtension extends BeforeAllCallback, AfterAllCallback, BeforeEachCallback, AfterEachCallback {
+
+ /**
+ * Gets the {@link CamelContext} created by this extension
+ * @return an instance of the Camel context to use in the test
+ */
+ CamelContext getContext();
+
+ /**
+ * Creates a {@link ProducerTemplate} from the context
+ * @return an instance of a producer template to use with the test
+ */
+ ProducerTemplate getProducerTemplate();
+
+ /**
+ * Creates a {@link ConsumerTemplate} from the context
+ * @return an instance of a consumer template to use with the test
+ */
+ ConsumerTemplate getConsumerTemplate();
+
+ /**
+ * Gets a {@link MockEndpoint} for the given URI. If the endpoint does not exist, it will be created
+ * @param uri the URI to create the mock to
+ * @return a Mock endpoint instance for the given URI
+ */
+ MockEndpoint getMockEndpoint(String uri);
+
+ /**
+ * Gets a {@link MockEndpoint} for the given URI.
+ * @param uri the URI to create the mock to
+ * @param create whether to create the endpoint if it does not exist
+ * @return a Mock endpoint instance for the given URI
+ * @throws NoSuchEndpointException if the endpoint does not exist and a new one should not be created
+ */
+ MockEndpoint getMockEndpoint(String uri, boolean create) throws NoSuchEndpointException;
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextFixture.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextFixture.java
new file mode 100644
index 00000000000..b92f7136525
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextFixture.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation indicates the need for configuring Camel contexts. Use it to annotate methods that configure the context.
+ * <p>
+ * The signature for such methods should be: <code>public void configureContext(CamelContext context)</code>.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Target({ElementType.TYPE, ElementType.METHOD})
+public @interface ContextFixture {
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextLifeCycleManager.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextLifeCycleManager.java
new file mode 100644
index 00000000000..12805a015a7
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/ContextLifeCycleManager.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core;
+
+import org.apache.camel.CamelContext;
+
+/**
+ * A life cycle manager for the Camel instance manages how the context is started, stopped and configured based on which
+ * phase of the test is being executed.
+ */
+public interface ContextLifeCycleManager {
+
+ /**
+ * A hook to execute after all tests are executed
+ * @param context the context instance
+ */
+ void afterAll(CamelContext context);
+
+ /**
+ * A hook to execute before all tests are executed
+ * @param context the context instance
+ */
+ void beforeAll(CamelContext context);
+
+ /**
+ * A hook to execute after each test is executed
+ * @param context the context instance
+ */
+ void afterEach(CamelContext context);
+
+ /**
+ * A hook to execute before each test is executed
+ * @param context the context instance
+ */
+ void beforeEach(CamelContext context);
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultCamelContextExtension.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultCamelContextExtension.java
new file mode 100644
index 00000000000..7063ab210b1
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultCamelContextExtension.java
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.core;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.Arrays;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.NoSuchEndpointException;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.util.URISupport;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * A simple Camel context extension suitable for most of the simple use cases in Camel and end-user applications.
+ */
+public class DefaultCamelContextExtension implements CamelContextExtension {
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultCamelContextExtension.class);
+ private final ContextLifeCycleManager lifeCycleManager;
+ private CamelContext context;
+ private ProducerTemplate producerTemplate;
+ private ConsumerTemplate consumerTemplate;
+
+ /**
+ * Creates a new instance of the extension
+ */
+ public DefaultCamelContextExtension() {
+ this(new DefaultContextLifeCycleManager());
+ }
+
+ /**
+ * Creates a new instance of the extension with a custom {@link ContextLifeCycleManager}
+ *
+ * @param lifeCycleManager a life cycle manager for the context
+ */
+ public DefaultCamelContextExtension(ContextLifeCycleManager lifeCycleManager) {
+ this.lifeCycleManager = lifeCycleManager;
+ }
+
+ /**
+ * Resolves an endpoint and asserts that it is found.
+ */
+ public static <T extends Endpoint> T resolveMandatoryEndpoint(CamelContext context, String endpointUri, Class<T> endpointType) {
+ T endpoint = context.getEndpoint(endpointUri, endpointType);
+
+ assertNotNull(endpoint, "No endpoint found for URI: " + endpointUri);
+
+ return endpoint;
+ }
+
+ private static String commonFixtureMessage(Class<? extends Annotation> annotationClass, Object instance) {
+ return "Unable to setup fixture " + annotationClass.getSimpleName() + " on " + instance.getClass().getName();
+ }
+
+ private static String invocationTargetMessage(Class<? extends Annotation> annotationClass, Object instance, String methodName) {
+ return commonFixtureMessage(annotationClass, instance) + " due to invocation target exception to method: " + methodName;
+ }
+
+ private static String illegalAccessMessage(Class<? extends Annotation> annotationClass, Object instance, String methodName) {
+ return commonFixtureMessage(annotationClass, instance) + " due to illegal access to method: " + methodName;
+ }
+
+ protected CamelContext createCamelContext() {
+ return new DefaultCamelContext();
+ }
+
+ @Override
+ public void afterAll(ExtensionContext extensionContext) throws Exception {
+ lifeCycleManager.afterAll(context);
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext extensionContext) throws Exception {
+ context = createCamelContext();
+
+ producerTemplate = context.createProducerTemplate();
+ producerTemplate.start();
+
+ consumerTemplate = context.createConsumerTemplate();
+ consumerTemplate.start();
+
+ lifeCycleManager.beforeAll(context);
+ }
+
+ @Override
+ public void beforeEach(ExtensionContext extensionContext) throws Exception {
+ final Object o = extensionContext.getTestInstance().get();
+
+ LOG.info("********************************************************************************");
+ LOG.info("Testing: {} ({})", extensionContext.getDisplayName(), o.getClass().getName());
+
+ setupFixtureFields(extensionContext, EndpointInject.class, o);
+ setupFixtureFields(extensionContext, BindToRegistry.class, o);
+
+ if (!context.isStarted()) {
+ setupFixture(extensionContext, ContextFixture.class, o);
+ setupFixture(extensionContext, RouteFixture.class, o);
+
+ lifeCycleManager.beforeEach(context);
+ }
+ }
+
+ @Override
+ public void afterEach(ExtensionContext extensionContext) throws Exception {
+ lifeCycleManager.afterEach(context);
+
+ final Object o = extensionContext.getTestInstance().get();
+ LOG.info("Testing done: {} ({})", extensionContext.getDisplayName(), o.getClass().getName());
+ LOG.info("********************************************************************************");
+ }
+
+ private void setupFixture(ExtensionContext extensionContext, Class<? extends Annotation> annotationClass, Object instance) {
+ final Class<?> testClass = extensionContext.getTestClass().get();
+
+ Arrays.stream(testClass.getMethods()).filter(m -> m.isAnnotationPresent(annotationClass)).forEach(m -> doInvokeFixture(annotationClass, instance, m));
+ }
+
+ private void setupFixtureFields(ExtensionContext extensionContext, Class<? extends Annotation> annotationClass, Object instance) {
+ final Class<?> testClass = extensionContext.getTestClass().get();
+
+ var superClass = testClass.getSuperclass();
+ while (superClass != null) {
+ Arrays.stream(superClass.getDeclaredFields()).filter(m -> m.isAnnotationPresent(annotationClass)).forEach(f -> doInvokeFixture(f.getAnnotation(annotationClass), instance, f));
+
+ superClass = superClass.getSuperclass();
+ }
+
+ Arrays.stream(testClass.getDeclaredFields()).filter(f -> f.isAnnotationPresent(annotationClass)).forEach(f -> doInvokeFixture(f.getAnnotation(annotationClass), instance, f));
+ }
+
+ private void doInvokeFixture(Class<? extends Annotation> annotationClass, Object instance, Method m) {
+ var methodName = m.getName();
+ LOG.trace("Checking instance method: {}", methodName);
+ try {
+ m.invoke(instance, context);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(illegalAccessMessage(annotationClass, instance, methodName), e);
+ } catch (InvocationTargetException e) {
+ throw new RuntimeException(invocationTargetMessage(annotationClass, instance, methodName), e);
+ }
+ }
+
+ private void doInvokeFixture(Annotation annotation, Object instance, Field f) {
+ var fieldName = f.getName();
+ LOG.trace("Checking instance field: {}", fieldName);
+ try {
+ if (!Modifier.isPublic(f.getModifiers())) {
+ f.setAccessible(true);
+ }
+
+ if (annotation instanceof BindToRegistry r) {
+ String bindValue = r.value();
+
+ context.getRegistry().bind(bindValue, f.get(instance));
+
+ return;
+ }
+
+ if (annotation instanceof EndpointInject e) {
+ String uri = e.value();
+
+ if (f.getType() == MockEndpoint.class) {
+ final MockEndpoint mockEndpoint = getMockEndpoint(uri);
+
+ assert mockEndpoint != null;
+
+ f.set(instance, mockEndpoint);
+ } else {
+ final Endpoint endpoint = context.getEndpoint(uri);
+
+ assert endpoint != null;
+ f.set(instance, endpoint);
+ }
+ }
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(illegalAccessMessage(annotation.getClass(), instance, fieldName), e);
+ }
+ }
+
+ @Override
+ public CamelContext getContext() {
+ return context;
+ }
+
+ @Override
+ public ProducerTemplate getProducerTemplate() {
+ return producerTemplate;
+ }
+
+ @Override
+ public ConsumerTemplate getConsumerTemplate() {
+ return consumerTemplate;
+ }
+
+ /**
+ * Resolves a mandatory endpoint for the given URI and expected type or an exception is thrown
+ *
+ * @param uri the Camel <a href="">URI</a> to use to create or resolve an endpoint
+ * @param endpointType the endpoint type (i.e., its class) to resolve
+ * @return the endpoint
+ */
+ protected <T extends Endpoint> T resolveMandatoryEndpoint(String uri, Class<T> endpointType) {
+ return resolveMandatoryEndpoint(context, uri, endpointType);
+ }
+
+ @Override
+ public MockEndpoint getMockEndpoint(String uri) {
+ MockEndpoint mock = getMockEndpoint(uri, true);
+
+ return mock;
+ }
+
+ @Override
+ public MockEndpoint getMockEndpoint(String uri, boolean create) throws NoSuchEndpointException {
+ // look for existing mock endpoints that have the same queue name, and
+ // to
+ // do that we need to normalize uri and strip out query parameters and
+ // whatnot
+ String n;
+ try {
+ n = URISupport.normalizeUri(uri);
+ } catch (Exception e) {
+ throw RuntimeCamelException.wrapRuntimeException(e);
+ }
+ // strip query
+ int idx = n.indexOf('?');
+ if (idx != -1) {
+ n = n.substring(0, idx);
+ }
+ final String target = n;
+
+ // lookup endpoints in registry and try to find it
+ MockEndpoint found = (MockEndpoint) context.getEndpointRegistry().values().stream().filter(e -> e instanceof MockEndpoint).filter(e -> {
+ String t = e.getEndpointUri();
+ // strip query
+ int idx2 = t.indexOf('?');
+ if (idx2 != -1) {
+ t = t.substring(0, idx2);
+ }
+ return t.equals(target);
+ }).findFirst().orElse(null);
+
+ if (found != null) {
+ return found;
+ }
+
+ if (create) {
+ return resolveMandatoryEndpoint(uri, MockEndpoint.class);
+ } else {
+ throw new NoSuchEndpointException(String.format("MockEndpoint %s does not exist.", uri));
+ }
+ }
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultContextLifeCycleManager.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultContextLifeCycleManager.java
new file mode 100644
index 00000000000..7c395a473ea
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/DefaultContextLifeCycleManager.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.mock.MockEndpoint;
+
+/**
+ * A default lifecycle manager suitable for most of the tests within Camel and end-user applications
+ */
+public class DefaultContextLifeCycleManager implements ContextLifeCycleManager {
+ public static final int DEFAULT_SHUTDOWN_TIMEOUT = 10;
+ private int shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
+ private boolean reset = true;
+
+ /**
+ * Creates a new instance of this class
+ */
+ public DefaultContextLifeCycleManager() {
+ }
+
+ /**
+ * Creates a new instance of this class
+ * @param shutdownTimeout the shutdown timeout
+ * @param reset whether to reset any {@link MockEndpoint} after each test execution
+ */
+ public DefaultContextLifeCycleManager(int shutdownTimeout, boolean reset) {
+ this.shutdownTimeout = shutdownTimeout;
+ this.reset = reset;
+ }
+
+ @Override
+ public void afterAll(CamelContext context) {
+ context.shutdown();
+ }
+
+ @Override
+ public void beforeAll(CamelContext context) {
+ context.getShutdownStrategy().setTimeout(shutdownTimeout);
+ }
+
+ @Override
+ public void afterEach(CamelContext context) {
+ if (reset) {
+ MockEndpoint.resetMocks(context);
+ }
+ }
+
+ @Override
+ public void beforeEach(CamelContext context) {
+ context.start();
+ }
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/RouteFixture.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/RouteFixture.java
new file mode 100644
index 00000000000..a37cad638b7
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/RouteFixture.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation indicates the need for configuring Camel routes to a context. Use it to annotate methods that configure the routes.
+ *
+ * The signature for such methods should be: <code>public void configureRoute(CamelContext context)</code>.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Target({ElementType.TYPE, ElementType.METHOD})
+public @interface RouteFixture {
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableContext.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableContext.java
new file mode 100644
index 00000000000..b837f62ed86
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableContext.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core.api;
+
+import org.apache.camel.CamelContext;
+
+/**
+ * A utility interface to indicate classes that configure the context. Usage of this interface is optional, but
+ * recommended for consistency.
+ */
+public interface ConfigurableContext {
+
+ /**
+ * Configures the context
+ * @param context the context to configure
+ * @throws Exception if the context cannot be configured
+ */
+ void configureContext(CamelContext context) throws Exception;
+
+}
diff --git a/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableRoute.java b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableRoute.java
new file mode 100644
index 00000000000..25306752041
--- /dev/null
+++ b/test-infra/camel-test-infra-core/src/test/java/org/apache/camel/test/infra/core/api/ConfigurableRoute.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.test.infra.core.api;
+
+import org.apache.camel.CamelContext;
+
+/**
+ * A utility interface to indicate classes that configure the routes used by the context. Usage of this interface is optional, but
+ * recommended for consistency.
+ */
+public interface ConfigurableRoute {
+
+ /**
+ * Creates the route builder and add it to the context
+ * @param context the context to configure
+ * @throws Exception if the context cannot be configured
+ */
+ void createRouteBuilder(CamelContext context) throws Exception;
+}
diff --git a/test-infra/pom.xml b/test-infra/pom.xml
index 84331f99481..21b6f2ca2c2 100644
--- a/test-infra/pom.xml
+++ b/test-infra/pom.xml
@@ -75,5 +75,6 @@
<module>camel-test-infra-hashicorp-vault</module>
<module>camel-test-infra-jetty</module>
<module>camel-test-infra-etcd3</module>
+ <module>camel-test-infra-core</module>
</modules>
</project>
[camel] 04/04: (chores) camel-test-infra-kafka: improve logging when starting up the service fails
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit e7825a48c9f3d1202333c4f311330be55ff30257
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Tue Jan 24 16:03:06 2023 +0100
(chores) camel-test-infra-kafka: improve logging when starting up the service fails
---
.../camel/test/infra/kafka/services/KafkaService.java | 14 +++++++++++++-
1 file changed, 13 insertions(+), 1 deletion(-)
diff --git a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaService.java b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaService.java
index 92c376546df..57baa6311c8 100644
--- a/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaService.java
+++ b/test-infra/camel-test-infra-kafka/src/test/java/org/apache/camel/test/infra/kafka/services/KafkaService.java
@@ -23,6 +23,8 @@ import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Provides an interface for any type of Kafka service: remote instances, local container, etc
@@ -39,7 +41,17 @@ public interface KafkaService
@Override
default void beforeAll(ExtensionContext extensionContext) throws Exception {
- initialize();
+ try {
+ initialize();
+ } catch (Exception e) {
+ Logger log = LoggerFactory.getLogger(KafkaService.class);
+
+ final Object o = extensionContext.getTestInstance().get();
+ log.error("Failed to initialize service {} for test {} on ({})", this.getClass().getSimpleName(),
+ extensionContext.getDisplayName(), o.getClass().getName());
+
+ throw e;
+ }
}
@Override
[camel] 03/04: CAMEL-18957: convert camel-kafka to the new camel-test-infra-core
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit e26777ab82e65297584314fa400cf44d0c6cb5b9
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Thu Jan 19 18:46:33 2023 +0100
CAMEL-18957: convert camel-kafka to the new camel-test-infra-core
---
components/camel-kafka/pom.xml | 12 ++
.../kafka/KafkaIdempotentRepository.java | 2 +
.../camel/component/kafka/KafkaAutowireTest.java | 14 +-
.../camel/component/kafka/KafkaComponentTest.java | 18 +-
.../integration/BaseEmbeddedKafkaTestSupport.java | 55 +++--
.../KafkaConsumerAsyncManualCommitIT.java | 39 ++--
.../kafka/integration/KafkaConsumerAuthIT.java | 49 ++++-
.../integration/KafkaConsumerAuthInvalidIT.java | 58 ++++--
.../KafkaConsumerAuthInvalidWithReconnectIT.java | 63 +++---
.../integration/KafkaConsumerAuthManualTest.java | 30 ++-
...KafkaConsumerAutoInstResumeRouteStrategyIT.java | 22 +-
.../integration/KafkaConsumerBatchSizeIT.java | 20 +-
.../kafka/integration/KafkaConsumerFullIT.java | 58 +++---
.../integration/KafkaConsumerHealthCheckIT.java | 227 ---------------------
.../integration/KafkaConsumerIdempotentIT.java | 36 ++--
...kaConsumerIdempotentWithCustomSerializerIT.java | 35 ++--
.../KafkaConsumerIdempotentWithProcessorIT.java | 33 ++-
.../KafkaConsumerLastRecordHeaderIT.java | 6 +-
.../integration/KafkaConsumerRebalanceIT.java | 19 +-
.../kafka/integration/KafkaConsumerStopIT.java | 30 ++-
.../integration/KafkaConsumerTopicIsPatternIT.java | 18 +-
.../kafka/integration/KafkaProducerFullIT.java | 212 +++++++++----------
.../component/kafka/integration/KafkaToDIT.java | 7 +-
.../kafka/integration/KafkaTransactionIT.java | 82 +++-----
.../{ => commit}/BaseManualCommitTestSupport.java | 13 +-
.../commit/KafkaConsumerAsyncCommitIT.java | 28 +--
.../KafkaConsumerAsyncWithOffsetRepoCommitIT.java | 28 +--
.../commit/KafkaConsumerNoopCommitIT.java | 30 +--
.../commit/KafkaConsumerSyncCommitIT.java | 17 +-
.../KafkaConsumerSyncWithOffsetRepoCommitIT.java | 31 ++-
.../KafkaAdminUtil.java} | 70 ++-----
.../KafkaTestUtil.java} | 34 ++-
.../kafka/integration/common/TestProducerUtil.java | 48 +++++
.../health/KafkaConsumerBadPortHealthCheckIT.java | 110 +++-------
...fkaConsumerBadPortSupervisingHealthCheckIT.java | 140 ++++---------
.../health/KafkaConsumerHealthCheckIT.java | 188 +++++++++++++++++
.../KafkaConsumerUnresolvableHealthCheckIT.java | 143 ++++---------
.../health/KafkaHealthCheckTestSupport.java | 103 ++++++++++
.../KafkaPausableConsumerCircuitBreakerIT.java | 75 +++----
.../integration/pause/KafkaPausableConsumerIT.java | 107 +++++-----
.../kafka/KafkaIdempotentRepositoryEagerIT.java | 39 ++--
.../kafka/KafkaIdempotentRepositoryNonEagerIT.java | 63 ++++--
.../KafkaIdempotentRepositoryPersistenceIT.java | 34 +--
.../idempotent/kafka/SimpleIdempotentTest.java | 47 +++++
44 files changed, 1223 insertions(+), 1270 deletions(-)
diff --git a/components/camel-kafka/pom.xml b/components/camel-kafka/pom.xml
index 9404f7b857d..a3df7100dc9 100644
--- a/components/camel-kafka/pom.xml
+++ b/components/camel-kafka/pom.xml
@@ -68,6 +68,14 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test-infra-kafka</artifactId>
@@ -95,6 +103,10 @@
<scope>test</scope>
</dependency>
+ <!--
+ It does not really have a dependency on camel-test-junit5, but it brings a lot of dependencies that are needed for
+ running the tests
+ -->
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test-junit5</artifactId>
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
index 8402d955e2a..20de5f0128e 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
@@ -359,6 +359,8 @@ public class KafkaIdempotentRepository extends ServiceSupport implements Idempot
private void broadcastAction(String key, CacheAction action) {
try {
log.debug("Broadcasting action:{} for key:{}", action, key);
+ assert producer != null;
+
producer.send(new ProducerRecord<>(topic, key, action.toString())).get(); // sync
// send
} catch (ExecutionException | InterruptedException e) {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java
index 81bf1fbf5b2..c629f7aaeb5 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java
@@ -17,15 +17,23 @@
package org.apache.camel.component.kafka;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.camel.CamelContext;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertSame;
-public class KafkaAutowireTest extends CamelTestSupport {
+public class KafkaAutowireTest {
+
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+
+ private CamelContext context = contextExtension.getContext();
@BindToRegistry
- KafkaClientFactory clientFactory = new TestKafkaClientFactory();
+ private KafkaClientFactory clientFactory = new TestKafkaClientFactory();
@Test
public void testKafkaComponentAutowiring() {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
index f01da289f0c..97147d48096 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
@@ -20,20 +20,34 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+import org.apache.camel.CamelContext;
import org.apache.camel.support.jsse.KeyStoreParameters;
import org.apache.camel.support.jsse.SSLContextParameters;
import org.apache.camel.support.jsse.TrustManagersParameters;
-import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
-public class KafkaComponentTest extends CamelTestSupport {
+public class KafkaComponentTest {
+
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+
+ private CamelContext context = contextExtension.getContext();
+
+ @AfterEach
+ void clear() {
+ context.removeComponent("kafka");
+ }
@Test
public void testPropertiesSet() {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
index 02e27dbefb2..c6bc92a6cba 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaTestSupport.java
@@ -19,46 +19,67 @@ package org.apache.camel.component.kafka.integration;
import java.util.Properties;
import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.component.kafka.integration.common.KafkaAdminUtil;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
+import org.apache.camel.test.infra.core.api.ConfigurableRoute;
import org.apache.camel.test.infra.kafka.services.KafkaService;
import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
import org.apache.kafka.clients.admin.AdminClient;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.extension.RegisterExtension;
-public abstract class BaseEmbeddedKafkaTestSupport extends AbstractKafkaTestSupport {
+public abstract class BaseEmbeddedKafkaTestSupport implements ConfigurableRoute {
+ @Order(1)
@RegisterExtension
- public static KafkaService service = KafkaServiceFactory.createSingletonService();
+ protected static KafkaService service = KafkaServiceFactory.createSingletonService();
+
+ @Order(2)
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
protected static AdminClient kafkaAdminClient;
- @BeforeAll
- public static void beforeClass() {
- AbstractKafkaTestSupport.setServiceProperties(service);
+ @BeforeEach
+ public void beforeClass() {
+ KafkaTestUtil.setServiceProperties(service);
}
@BeforeEach
public void setKafkaAdminClient() {
if (kafkaAdminClient == null) {
- kafkaAdminClient = createAdminClient();
+ kafkaAdminClient = KafkaAdminUtil.createAdminClient(service);
}
}
- protected Properties getDefaultProperties() {
- return getDefaultProperties(service);
- }
+ @ContextFixture
+ public void configureKafka(CamelContext context) {
+ context.getPropertiesComponent().setLocation("ref:prop");
- @Override
- protected CamelContext createCamelContext() throws Exception {
- return createCamelContextFromService(service);
+ KafkaComponent kafka = new KafkaComponent(context);
+ kafka.init();
+ kafka.getConfiguration().setBrokers(service.getBootstrapServers());
+ context.addComponent("kafka", kafka);
}
- protected static String getBootstrapServers() {
- return service.getBootstrapServers();
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
}
- private static AdminClient createAdminClient() {
- return createAdminClient(service);
+ protected abstract RouteBuilder createRouteBuilder();
+
+ protected Properties getDefaultProperties() {
+ return KafkaTestUtil.getDefaultProperties(service);
}
+ protected static String getBootstrapServers() {
+ return service.getBootstrapServers();
+ }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
index 9f01dfd3540..06e6b5ac1e2 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
@@ -21,8 +21,7 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.builder.AggregationStrategies;
import org.apache.camel.builder.RouteBuilder;
@@ -30,6 +29,7 @@ import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
import org.apache.camel.component.kafka.consumer.KafkaManualCommitFactory;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.awaitility.Awaitility;
@@ -47,33 +47,23 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-
public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "testManualCommitTest";
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAsyncManualCommitIT.class);
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerAsyncManualCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
- + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#testFactory")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- @EndpointInject("mock:resultBar")
- private MockEndpoint toBar;
-
@BindToRegistry("testFactory")
private KafkaManualCommitFactory manualCommitFactory = new DefaultKafkaManualAsyncCommitFactory();
+ private CamelContext context = contextExtension.getContext();
+
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
private volatile int failCount;
@BeforeEach
public void before() {
- Properties props = getDefaultProperties();
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
}
@@ -84,17 +74,20 @@ public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSuppo
}
}
- @Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("foo").to("direct:aggregate");
+ String uri = "kafka:" + TOPIC + "?brokers=" + service.getBootstrapServers()
+ + "&groupId=KafkaConsumerAsyncManualCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#testFactory";
+
+ from(uri).routeId("foo").to("direct:aggregate");
// With sync manual commit, this would throw a concurrent modification exception
// It can be used in aggregator with completion timeout/interval for instance
// WARN: records from one partition must be processed by one unique thread
- from("direct:aggregate").routeId("aggregate").to(to)
+ from("direct:aggregate").routeId("aggregate").to(KafkaTestUtil.MOCK_RESULT)
.aggregate()
.constant(true)
.completionTimeout(1)
@@ -112,7 +105,7 @@ public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSuppo
failCount++;
}
});
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ from(uri).routeId("bar").autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
@@ -121,6 +114,8 @@ public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSuppo
@Order(1)
@Test
void testLastRecordBeforeCommitHeader() {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
@@ -142,7 +137,7 @@ public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSuppo
@Order(2)
@Test
void kafkaManualCommit() throws Exception {
- to.reset();
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
// Second step: We shut down our route, we expect nothing will be recovered by our route
context.getRouteController().stopRoute("foo");
@@ -161,10 +156,10 @@ public class KafkaConsumerAsyncManualCommitIT extends BaseEmbeddedKafkaTestSuppo
@Order(3)
@Test
void testResumeFromTheRightPoint() throws Exception {
- to.reset();
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
// Fourth step: We start again our route, since we have been committing the offsets from the first step,
- // we will expect to consume from the latest committed offset e.g from offset 5
+ // we will expect to consume from the latest committed offset i.e. from offset 5
context.getRouteController().startRoute("foo");
to.expectedMessageCount(3);
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
index 77fb886caf5..487dfc5fef9 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthIT.java
@@ -22,12 +22,18 @@ import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.stream.StreamSupport;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaAdminUtil;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
+import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -37,9 +43,9 @@ import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,20 +55,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
+public class KafkaConsumerAuthIT {
public static final String TOPIC = "test-auth-full";
- private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthIT.class);
+ @Order(1)
+ @RegisterExtension
+ public static ContainerLocalAuthKafkaService service = new ContainerLocalAuthKafkaService("/kafka-jaas.config");
- @EndpointInject("mock:result")
- private MockEndpoint to;
+ @Order(2)
+ @RegisterExtension
+ public static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+
+ protected static AdminClient kafkaAdminClient;
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthIT.class);
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
public void before() {
- Properties props = getDefaultProperties();
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
+
props.put(SaslConfigs.SASL_JAAS_CONFIG,
ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret"));
props.put("security.protocol", "SASL_PLAINTEXT");
@@ -77,6 +90,13 @@ public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
MockConsumerInterceptor.recordsCaptured.clear();
}
+ @BeforeEach
+ public void setKafkaAdminClient() {
+ if (kafkaAdminClient == null) {
+ kafkaAdminClient = KafkaAdminUtil.createAdminClient(service);
+ }
+ }
+
@AfterEach
public void after() {
if (producer != null) {
@@ -86,7 +106,11 @@ public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
}
- @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -96,14 +120,15 @@ public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
= ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret");
fromF("kafka:%s"
- + "?groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "?brokers=%s&groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer&clientId=camel-kafka-auth-test"
+ "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=%s"
+ "&saslMechanism=PLAIN&securityProtocol=SASL_PLAINTEXT&saslJaasConfig=%s", TOPIC,
+ service.getBootstrapServers(),
"KafkaConsumerAuthIT", "org.apache.camel.component.kafka.MockConsumerInterceptor", simpleSaslJaasConfig)
.process(
exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("full-it").to(to);
+ .routeId("full-it").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -113,6 +138,8 @@ public class KafkaConsumerAuthIT extends BaseEmbeddedKafkaAuthTestSupport {
@Order(1)
@Test
public void kafkaMessageIsConsumedByCamel() throws InterruptedException, ExecutionException {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
String propagatedHeaderKey = "PropagatedCustomHeader";
byte[] propagatedHeaderValue = "propagated header value".getBytes();
String skippedHeaderKey = "CamelSkippedHeader";
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidIT.java
index 902cad81ff4..f8a72007dd1 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidIT.java
@@ -20,13 +20,21 @@ import java.util.Collections;
import java.util.Map;
import java.util.Properties;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.ExtendedCamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaAdminUtil;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
+import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.common.config.SaslConfigs;
import org.junit.Assert;
@@ -35,11 +43,12 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,23 +57,25 @@ import static org.junit.jupiter.api.Assertions.fail;
@DisabledIfSystemProperty(named = "ci.env.name", matches = "github.com", disabledReason = "Flaky on Github CI")
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport {
+public class KafkaConsumerAuthInvalidIT {
public static final String TOPIC = "test-auth-invalid-it";
- private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthInvalidIT.class);
+ protected static AdminClient kafkaAdminClient;
- @EndpointInject("mock:result")
- private MockEndpoint to;
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthInvalidIT.class);
- @EndpointInject("mock:dlq")
- private MockEndpoint dlq;
+ @Order(1)
+ @RegisterExtension
+ private static KafkaService service = KafkaServiceFactory.createSingletonService();
+ @Order(2)
+ @RegisterExtension
+ private static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
public void before() {
- Properties props = getDefaultProperties();
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
props.put(SaslConfigs.SASL_JAAS_CONFIG,
ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret"));
props.put("security.protocol", "SASL_PLAINTEXT");
@@ -79,6 +90,13 @@ public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport
MockConsumerInterceptor.recordsCaptured.clear();
}
+ @BeforeEach
+ public void setKafkaAdminClient() {
+ if (kafkaAdminClient == null) {
+ kafkaAdminClient = KafkaTestUtil.createAdminClient(service);
+ }
+ }
+
@AfterEach
public void after() {
if (producer != null) {
@@ -88,7 +106,11 @@ public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
}
- @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -99,17 +121,17 @@ public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport
getCamelContext().adapt(ExtendedCamelContext.class)
.setErrorHandlerFactory(
- deadLetterChannel(dlq));
+ deadLetterChannel("mock:dlq"));
fromF("kafka:%s"
- + "?groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "?brokers=%s&groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
+ "&saslMechanism=PLAIN&securityProtocol=SASL_PLAINTEXT&saslJaasConfig=%s", TOPIC,
- "KafkaConsumerAuthInvalidIT", simpleSaslJaasConfig)
+ service.getBootstrapServers(), "KafkaConsumerAuthInvalidIT", simpleSaslJaasConfig)
.process(
exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("should-no-work").to(to);
+ .routeId("should-no-work").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -118,6 +140,9 @@ public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport
@Timeout(30)
@Test
public void kafkaMessageIsConsumedByCamel() throws InterruptedException {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+ MockEndpoint dlq = contextExtension.getMockEndpoint("mock:dlq");
+
dlq.expectedMessageCount(1);
dlq.assertIsSatisfied(3000);
@@ -125,7 +150,8 @@ public class KafkaConsumerAuthInvalidIT extends BaseEmbeddedKafkaAuthTestSupport
to.assertIsSatisfied(3000);
final Map<String, ConsumerGroupDescription> allGroups
- = assertDoesNotThrow(() -> getConsumerGroupInfo("KafkaConsumerAuthInvalidIT"));
+ = assertDoesNotThrow(() -> KafkaAdminUtil.getConsumerGroupInfo("KafkaConsumerAuthInvalidIT",
+ kafkaAdminClient));
final ConsumerGroupDescription groupInfo = allGroups.get("KafkaConsumerAuthInvalidIT");
Assert.assertEquals("There should be no members in this group", 0, groupInfo.members().size());
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidWithReconnectIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidWithReconnectIT.java
index 81521b4785e..76789b53076 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidWithReconnectIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthInvalidWithReconnectIT.java
@@ -16,18 +16,18 @@
*/
package org.apache.camel.component.kafka.integration;
-import java.util.Collections;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import org.apache.camel.CamelContext;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
-import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.kafka.integration.common.KafkaAdminUtil;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
@@ -40,8 +40,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,19 +49,15 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.fail;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSupport {
+public class KafkaConsumerAuthInvalidWithReconnectIT {
public static final String TOPIC = "test-auth-invalid-with-reconnect";
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAuthInvalidWithReconnectIT.class);
private static ContainerLocalAuthKafkaService service;
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- @EndpointInject("mock:dlq")
- private MockEndpoint dlq;
+ @RegisterExtension
+ private static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@@ -73,16 +69,7 @@ public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSu
@BeforeAll
public static void beforeClass() {
service.initialize();
- AbstractKafkaTestSupport.setServiceProperties(service);
- }
-
- protected Properties getDefaultProperties() {
- return getDefaultProperties(service);
- }
-
- @Override
- protected CamelContext createCamelContext() throws Exception {
- return createCamelContextFromService(service);
+ KafkaTestUtil.setServiceProperties(service);
}
protected static String getBootstrapServers() {
@@ -92,7 +79,7 @@ public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSu
@BeforeEach
public void before() {
- Properties props = getDefaultProperties();
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
props.put(SaslConfigs.SASL_JAAS_CONFIG,
ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret"));
props.put("security.protocol", "SASL_PLAINTEXT");
@@ -114,7 +101,11 @@ public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSu
}
}
- @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -123,31 +114,28 @@ public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSu
final String simpleSaslJaasConfig
= ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("camel", "camel-secret");
- fromF("kafka:%s"
- + "?groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ fromF("kafka:%s?brokers=%s"
+ + "&groupId=%s&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&pollOnError=RECONNECT"
+ "&saslMechanism=PLAIN&securityProtocol=SASL_PLAINTEXT&saslJaasConfig=%s", TOPIC,
+ service.getBootstrapServers(),
"KafkaConsumerAuthInvalidWithReconnectIT", simpleSaslJaasConfig)
.process(
exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("full-it").to(to);
+ .routeId("full-it").to(KafkaTestUtil.MOCK_DLQ);
}
};
}
- private Map<String, ConsumerGroupDescription> getConsumerGroupInfo(AdminClient adminClient, String groupId)
- throws InterruptedException, ExecutionException, TimeoutException {
- return adminClient.describeConsumerGroups(Collections.singletonList(groupId)).all().get(30, TimeUnit.SECONDS);
- }
-
@Test
@Order(1)
void testIsDisconnected() {
- AdminClient adminClient = BaseEmbeddedKafkaAuthTestSupport.createAuthAdminClient(service);
+ AdminClient adminClient = KafkaAdminUtil.createAuthAdminClient(service);
final Map<String, ConsumerGroupDescription> allGroups
- = assertDoesNotThrow(() -> getConsumerGroupInfo(adminClient, "KafkaConsumerAuthInvalidWithReconnectIT"));
+ = assertDoesNotThrow(
+ () -> KafkaAdminUtil.getConsumerGroupInfo("KafkaConsumerAuthInvalidWithReconnectIT", adminClient));
final ConsumerGroupDescription groupInfo = allGroups.get("KafkaConsumerAuthInvalidWithReconnectIT");
Assert.assertEquals("There should be no members in this group", 0, groupInfo.members().size());
@@ -166,14 +154,15 @@ public class KafkaConsumerAuthInvalidWithReconnectIT extends AbstractKafkaTestSu
service.initialize();
- AdminClient adminClient = BaseEmbeddedKafkaAuthTestSupport.createAuthAdminClient(service);
+ AdminClient adminClient = KafkaAdminUtil.createAuthAdminClient(service);
Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertIsConnected(adminClient));
adminClient.close();
}
private void assertIsConnected(AdminClient adminClient) {
final Map<String, ConsumerGroupDescription> allGroups
- = assertDoesNotThrow(() -> getConsumerGroupInfo(adminClient, "KafkaConsumerAuthInvalidWithReconnectIT"));
+ = assertDoesNotThrow(
+ () -> KafkaAdminUtil.getConsumerGroupInfo("KafkaConsumerAuthInvalidWithReconnectIT", adminClient));
Assert.assertTrue("There should be at least one group named KafkaConsumerAuthInvalidWithReconnectIT",
allGroups.size() >= 1);
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthManualTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthManualTest.java
index 1adf7563258..f45b5b9687f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthManualTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAuthManualTest.java
@@ -19,10 +19,14 @@ package org.apache.camel.component.kafka.integration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
import org.apache.kafka.common.config.SaslConfigs;
import org.awaitility.Awaitility;
@@ -36,6 +40,7 @@ import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.EnabledIfSystemProperties;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.fail;
@@ -47,8 +52,8 @@ import static org.junit.jupiter.api.Assertions.fail;
@EnabledIfSystemProperty(named = "kafka.manual.test.username", matches = ".*"),
@EnabledIfSystemProperty(named = "kafka.manual.test.password", matches = ".*"),
})
-public class KafkaConsumerAuthManualTest extends AbstractKafkaTestSupport {
- public static final String TOPIC = System.getProperty("kafka.manual.test.topic");
+public class KafkaConsumerAuthManualTest {
+ private static final String TOPIC = System.getProperty("kafka.manual.test.topic");
private static final String BOOTSTRAP_SERVERS = System.getProperty("bootstrapServers");
private static final String USERNAME = System.getProperty("kafka.manual.test.username");
private static final String PASSWORD = System.getProperty("kafka.manual.test.password");
@@ -56,15 +61,13 @@ public class KafkaConsumerAuthManualTest extends AbstractKafkaTestSupport {
private static final String SASL_MECHANISM = System.getProperty("kafka.manual.test.sasl.mechanism", "PLAIN");
private static final int MESSAGE_COUNT = Integer.valueOf(System.getProperty("kafka.manual.test.message.count", "5"));
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
+ @RegisterExtension
+ private static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+ private static volatile int receivedMessages;
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
- private volatile int receivedMessages;
- @Override
protected Properties getDefaultProperties() {
- Properties properties = AbstractKafkaTestSupport.getDefaultProperties(BOOTSTRAP_SERVERS);
+ Properties properties = KafkaTestUtil.getDefaultProperties(BOOTSTRAP_SERVERS);
properties.put(SaslConfigs.SASL_JAAS_CONFIG,
ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig(USERNAME, PASSWORD));
@@ -94,7 +97,11 @@ public class KafkaConsumerAuthManualTest extends AbstractKafkaTestSupport {
// clean all test topics
}
- @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -109,7 +116,7 @@ public class KafkaConsumerAuthManualTest extends AbstractKafkaTestSupport {
+ "&saslMechanism=%s&securityProtocol=%s&saslJaasConfig=%s", TOPIC, BOOTSTRAP_SERVERS,
"KafkaConsumerAuthManualTest", SASL_MECHANISM, SECURITY_PROTOCOL, simpleSaslJaasConfig)
.process(e -> receivedMessages++)
- .routeId("full-it").to(to);
+ .routeId("full-it").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -117,6 +124,7 @@ public class KafkaConsumerAuthManualTest extends AbstractKafkaTestSupport {
@DisplayName("Tests that Camel can adequately connect and consume from an authenticated remote Kafka instance")
@Test
public void kafkaMessageIsConsumedByCamel() throws InterruptedException {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAutoInstResumeRouteStrategyIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAutoInstResumeRouteStrategyIT.java
index 5710bc670ce..ccc141f27a3 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAutoInstResumeRouteStrategyIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAutoInstResumeRouteStrategyIT.java
@@ -20,10 +20,10 @@ import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
-import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.consumer.support.resume.KafkaResumable;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.resume.TransientResumeStrategy;
import org.apache.camel.processor.resume.kafka.KafkaResumeStrategyConfigurationBuilder;
@@ -33,16 +33,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class KafkaConsumerAutoInstResumeRouteStrategyIT extends BaseEmbeddedKafkaTestSupport {
- private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerAutoInstResumeRouteStrategyIT.class);
private static final String TOPIC = "resumable-route-auto";
- @EndpointInject("mock:result")
- private MockEndpoint result;
-
public static KafkaResumeStrategyConfigurationBuilder getDefaultKafkaResumeStrategyConfigurationBuilder() {
return KafkaResumeStrategyConfigurationBuilder.newBuilder()
.withBootstrapServers(service.getBootstrapServers())
@@ -58,7 +52,7 @@ public class KafkaConsumerAutoInstResumeRouteStrategyIT extends BaseEmbeddedKafk
@BeforeEach
public void before() {
- Properties props = getDefaultProperties();
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
KafkaProducer<Object, Object> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
@@ -66,16 +60,10 @@ public class KafkaConsumerAutoInstResumeRouteStrategyIT extends BaseEmbeddedKafk
}
}
- @Override
- protected void doPreSetup() throws Exception {
- super.doPreSetup();
-
- }
-
@Test
@Timeout(value = 30)
public void testOffsetIsBeingChecked() throws InterruptedException {
- MockEndpoint mock = getMockEndpoint("mock:result");
+ MockEndpoint mock = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
mock.expectedMessageCount(10);
mock.assertIsSatisfied();
@@ -86,7 +74,7 @@ public class KafkaConsumerAutoInstResumeRouteStrategyIT extends BaseEmbeddedKafk
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
- public void process(Exchange exchange) {
+ private void process(Exchange exchange) {
exchange.getMessage().setHeader(Exchange.OFFSET, KafkaResumable.of(exchange));
}
@@ -106,7 +94,7 @@ public class KafkaConsumerAutoInstResumeRouteStrategyIT extends BaseEmbeddedKafk
fromF("kafka:%s?groupId=%s_GROUP&autoCommitIntervalMs=1000", "resumable-route-auto-offsets",
"resumable-route-auto-offsets")
- .to("mock:result");
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerBatchSizeIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerBatchSizeIT.java
index 222981191f9..388a8b22d17 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerBatchSizeIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerBatchSizeIT.java
@@ -19,9 +19,9 @@ package org.apache.camel.component.kafka.integration;
import java.util.Collections;
import java.util.Properties;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
@@ -32,12 +32,6 @@ public class KafkaConsumerBatchSizeIT extends BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "test-batch";
- @EndpointInject("kafka:" + TOPIC + "?autoOffsetReset=earliest&autoCommitEnable=false&consumersCount=1")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
@@ -55,18 +49,20 @@ public class KafkaConsumerBatchSizeIT extends BaseEmbeddedKafkaTestSupport {
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
- @Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("foo").to(to).setId("First");
+ fromF("kafka:%s?brokers=%s&autoOffsetReset=earliest&autoCommitEnable=false&consumersCount=1",
+ TOPIC, service.getBootstrapServers())
+ .routeId("foo").to(KafkaTestUtil.MOCK_RESULT).setId("First");
}
};
}
@Test
public void kafkaMessagesIsConsumedByCamel() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
// First 2 must not be committed since batch size is 3
to.expectedBodiesReceivedInAnyOrder("m1", "m2");
@@ -82,7 +78,9 @@ public class KafkaConsumerBatchSizeIT extends BaseEmbeddedKafkaTestSupport {
to.expectedBodiesReceivedInAnyOrder("m3", "m4", "m5", "m6", "m7", "m8", "m9", "m10");
- // Restart endpoint,
+ // Restart endpoint
+ CamelContext context = contextExtension.getContext();
+
context.getRouteController().stopRoute("foo");
context.getRouteController().startRoute("foo");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
index a5756b4a4e8..598398462a7 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
@@ -22,15 +22,16 @@ import java.util.Properties;
import java.util.stream.StreamSupport;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.KafkaEndpoint;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
import org.apache.camel.component.kafka.SeekPolicy;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.junit.jupiter.api.AfterEach;
@@ -43,9 +44,9 @@ import org.junit.jupiter.api.TestMethodOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertTrue;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@@ -55,19 +56,14 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerFullIT.class);
- @BindToRegistry("myHeaderDeserializer")
- private MyKafkaHeaderDeserializer deserializer = new MyKafkaHeaderDeserializer();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerFullIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
+ private static final String FROM_URI = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerFullIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor";
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+ @BindToRegistry("myHeaderDeserializer")
+ private MyKafkaHeaderDeserializer bean = new MyKafkaHeaderDeserializer();
@BeforeEach
public void before() {
@@ -85,14 +81,18 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
}
- @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
-
@Override
public void configure() {
- from(from).process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("full-it").to(to);
+ from(FROM_URI)
+ .process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
+ .routeId("full-it").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -103,6 +103,9 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
String propagatedHeaderKey = "PropagatedCustomHeader";
byte[] propagatedHeaderValue = "propagated header value".getBytes();
String skippedHeaderKey = "CamelSkippedHeader";
+
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
// The LAST_RECORD_BEFORE_COMMIT header should not be configured on any
@@ -133,6 +136,8 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
public void kafkaRecordSpecificHeadersAreNotOverwritten() throws InterruptedException {
String propagatedHeaderKey = KafkaConstants.TOPIC;
byte[] propagatedHeaderValue = "propagated incorrect topic".getBytes();
+
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
to.expectedHeaderReceived(KafkaConstants.TOPIC, TOPIC);
ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", "message");
@@ -149,6 +154,7 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
@Test
@Order(1)
public void kafkaMessageIsConsumedByCamelSeekedToBeginning() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
for (int k = 0; k < 5; k++) {
@@ -164,10 +170,11 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
- // Restart endpoint,
+ // Restart endpoint
+ CamelContext context = contextExtension.getContext();
context.getRouteController().stopRoute("full-it");
- KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) from;
+ KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) context.getEndpoint(FROM_URI);
kafkaEndpoint.getConfiguration().setSeekTo(SeekPolicy.BEGINNING);
context.getRouteController().startRoute("full-it");
@@ -179,6 +186,8 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
@Order(4)
@Test
public void kafkaMessageIsConsumedByCamelSeekedToEnd() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
for (int k = 0; k < 5; k++) {
@@ -192,10 +201,11 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
to.expectedMessageCount(0);
- // Restart endpoint,
+ // Restart endpoint
+ CamelContext context = contextExtension.getContext();
context.getRouteController().stopRoute("full-it");
- KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) from;
+ KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) context.getEndpoint(FROM_URI);
kafkaEndpoint.getConfiguration().setSeekTo(SeekPolicy.END);
context.getRouteController().startRoute("full-it");
@@ -206,9 +216,11 @@ public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
@Order(5)
@Test
public void headerDeserializerCouldBeOverridden() {
+ CamelContext context = contextExtension.getContext();
+
KafkaEndpoint kafkaEndpoint
= context.getEndpoint("kafka:random_topic?headerDeserializer=#myHeaderDeserializer", KafkaEndpoint.class);
- assertIsInstanceOf(MyKafkaHeaderDeserializer.class, kafkaEndpoint.getConfiguration().getHeaderDeserializer());
+ assertInstanceOf(MyKafkaHeaderDeserializer.class, kafkaEndpoint.getConfiguration().getHeaderDeserializer());
}
private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
deleted file mode 100644
index e6fbcd24d6b..00000000000
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerHealthCheckIT.java
+++ /dev/null
@@ -1,227 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.component.kafka.integration;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.StreamSupport;
-
-import org.apache.camel.BindToRegistry;
-import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.kafka.KafkaComponent;
-import org.apache.camel.component.kafka.KafkaConstants;
-import org.apache.camel.component.kafka.MockConsumerInterceptor;
-import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.health.HealthCheck;
-import org.apache.camel.health.HealthCheckHelper;
-import org.apache.camel.health.HealthCheckRegistry;
-import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
-import org.apache.camel.test.infra.kafka.services.KafkaService;
-import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.header.internals.RecordHeader;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.MethodOrderer;
-import org.junit.jupiter.api.Order;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestMethodOrder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
-
-@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-
-public class KafkaConsumerHealthCheckIT extends CamelTestSupport {
- public static final String TOPIC = "test-health";
-
- public static KafkaService service = KafkaServiceFactory.createService();
-
- protected static AdminClient kafkaAdminClient;
-
- private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerHealthCheckIT.class);
-
- @BindToRegistry("myHeaderDeserializer")
- private MyKafkaHeaderDeserializer deserializer = new MyKafkaHeaderDeserializer();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
-
- @BeforeEach
- public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
- producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
- MockConsumerInterceptor.recordsCaptured.clear();
- }
-
- @BeforeAll
- public static void beforeClass() {
- service.initialize();
-
- LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
- System.setProperty("bootstrapServers", service.getBootstrapServers());
- System.setProperty("brokers", service.getBootstrapServers());
- }
-
- @AfterAll
- public static void afterClass() {
- service.shutdown();
- }
-
- @BeforeEach
- public void setKafkaAdminClient() {
- if (kafkaAdminClient == null) {
- kafkaAdminClient = AbstractKafkaTestSupport.createAdminClient(service);
- }
- }
-
- @AfterEach
- public void after() {
- if (producer != null) {
- producer.close();
- }
- // clean all test topics
- kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
- }
-
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext context = super.createCamelContext();
- context.getPropertiesComponent().setLocation("ref:prop");
-
- KafkaComponent kafka = new KafkaComponent(context);
- kafka.init();
- kafka.getConfiguration().setBrokers(service.getBootstrapServers());
- context.addComponent("kafka", kafka);
-
- // install health check manually (yes a bit cumbersome)
- HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
- registry.setCamelContext(context);
- Object hc = registry.resolveById("context");
- registry.register(hc);
- hc = registry.resolveById("routes");
- registry.register(hc);
- hc = registry.resolveById("consumers");
- registry.register(hc);
- context.setExtension(HealthCheckRegistry.class, registry);
-
- return context;
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
-
- @Override
- public void configure() {
- from(from)
- .process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("test-health-it").to(to);
- }
- };
- }
-
- @Order(1)
- @Test
- public void kafkaConsumerHealthCheck() throws InterruptedException {
- // health-check liveness should be UP
- Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
- boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
- Assertions.assertTrue(up, "liveness check");
-
- // health-check readiness should be ready
- await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
- Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
- boolean up2 = res2.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
- Assertions.assertTrue(up2, "readiness check");
- });
-
- String propagatedHeaderKey = "PropagatedCustomHeader";
- byte[] propagatedHeaderValue = "propagated header value".getBytes();
- String skippedHeaderKey = "CamelSkippedHeader";
- to.expectedMessageCount(5);
- to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
- // The LAST_RECORD_BEFORE_COMMIT header should not be configured on any
- // exchange because autoCommitEnable=true
- to.expectedHeaderValuesReceivedInAnyOrder(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, null, null, null, null, null);
- to.expectedHeaderReceived(propagatedHeaderKey, propagatedHeaderValue);
-
- for (int k = 0; k < 5; k++) {
- String msg = "message-" + k;
- ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
- data.headers().add(new RecordHeader("CamelSkippedHeader", "skipped header value".getBytes()));
- data.headers().add(new RecordHeader(propagatedHeaderKey, propagatedHeaderValue));
- producer.send(data);
- }
-
- to.assertIsSatisfied(3000);
-
- assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
- .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
-
- Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
- assertFalse(headers.containsKey(skippedHeaderKey), "Should not receive skipped header");
- assertTrue(headers.containsKey(propagatedHeaderKey), "Should receive propagated header");
-
- // and shutdown kafka which will make readiness report as DOWN
- service.shutdown();
-
- // health-check liveness should be UP
- res = HealthCheckHelper.invokeLiveness(context);
- up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
- Assertions.assertTrue(up, "liveness check");
- // but health-check readiness should NOT be ready
- await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
- Collection<HealthCheck.Result> res2 = HealthCheckHelper.invoke(context);
- Optional<HealthCheck.Result> down
- = res2.stream().filter(r -> r.getState().equals(HealthCheck.State.DOWN)).findFirst();
- Assertions.assertTrue(down.isPresent());
- String msg = down.get().getMessage().get();
- Assertions.assertTrue(msg.contains("KafkaConsumer is not ready"));
- Map<String, Object> map = down.get().getDetails();
- Assertions.assertEquals(TOPIC, map.get("topic"));
- Assertions.assertEquals("test-health-it", map.get("route.id"));
- });
- }
-
- private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
- }
-}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
index 682774dc765..5874a59b3cf 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
@@ -19,9 +19,8 @@ package org.apache.camel.component.kafka.integration;
import java.util.Arrays;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.junit.jupiter.api.AfterEach;
@@ -33,27 +32,15 @@ import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import static org.apache.camel.component.kafka.serde.KafkaSerdeHelper.numericHeader;
@DisabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests", matches = "false")
-class KafkaConsumerIdempotentIT extends KafkaConsumerIdempotentTestSupport {
+public class KafkaConsumerIdempotentIT extends KafkaConsumerIdempotentTestSupport {
public static final String TOPIC = "idempt";
- @BindToRegistry("kafkaIdempotentRepository")
- private KafkaIdempotentRepository kafkaIdempotentRepository
- = new KafkaIdempotentRepository("TEST_IDEMPOTENT", getBootstrapServers());
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerIdempotentIT&autoOffsetReset=earliest"
- + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
- + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
private int size = 200;
+ @BindToRegistry("kafkaIdempotentRepository")
+ private KafkaIdempotentRepository testIdempotent = new KafkaIdempotentRepository("TEST_IDEMPOTENT", getBootstrapServers());
+
@BeforeEach
public void before() {
kafkaAdminClient.deleteTopics(Arrays.asList(TOPIC, "TEST_IDEMPOTENT")).all();
@@ -65,17 +52,20 @@ class KafkaConsumerIdempotentIT extends KafkaConsumerIdempotentTestSupport {
kafkaAdminClient.deleteTopics(Arrays.asList(TOPIC, "TEST_IDEMPOTENT")).all();
}
- @Override
protected RouteBuilder createRouteBuilder() {
-
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("foo")
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerIdempotentIT&autoOffsetReset=earliest"
+ + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor").routeId("foo")
.idempotentConsumer(numericHeader("id"))
.idempotentRepository("kafkaIdempotentRepository")
- .to(to);
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -83,6 +73,8 @@ class KafkaConsumerIdempotentIT extends KafkaConsumerIdempotentTestSupport {
@Test
@DisplayName("Numeric headers is consumable when using idempotent (CAMEL-16914)")
void kafkaIdempotentMessageIsConsumedByCamel() {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
doRun(to, size);
}
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
index f50f90c2ba7..66270b49ecc 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
@@ -19,40 +19,26 @@ package org.apache.camel.component.kafka.integration;
import java.util.Arrays;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-class KafkaConsumerIdempotentWithCustomSerializerIT extends KafkaConsumerIdempotentTestSupport {
+public class KafkaConsumerIdempotentWithCustomSerializerIT extends KafkaConsumerIdempotentTestSupport {
public static final String TOPIC = "idempt2";
+ private int size = 200;
+
@BindToRegistry("kafkaIdempotentRepository")
private KafkaIdempotentRepository kafkaIdempotentRepository
= new KafkaIdempotentRepository("TEST_IDEMPOTENT", getBootstrapServers());
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerIdempotentWithCustomSerializerIT&autoOffsetReset=earliest"
- + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&headerDeserializer=#class:org.apache.camel.component.kafka.integration.CustomHeaderDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
- + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- private int size = 200;
-
@BeforeEach
public void before() {
- kafkaIdempotentRepository.clear();
kafkaAdminClient.deleteTopics(Arrays.asList(TOPIC, "TEST_IDEMPOTENT")).all();
doSend(size, TOPIC);
}
@@ -62,22 +48,29 @@ class KafkaConsumerIdempotentWithCustomSerializerIT extends KafkaConsumerIdempot
kafkaAdminClient.deleteTopics(Arrays.asList(TOPIC, "TEST_IDEMPOTENT")).all();
}
- @Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("foo")
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerIdempotentWithCustomSerializerIT&autoOffsetReset=earliest"
+ + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&headerDeserializer=#class:org.apache.camel.component.kafka.integration.CustomHeaderDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor").routeId("foo")
.idempotentConsumer(header("id"))
.idempotentRepository("kafkaIdempotentRepository")
- .to(to);
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@Test
void kafkaMessageIsConsumedByCamel() {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
doRun(to, size);
}
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
index 2f270b99fab..467996bb2e6 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
@@ -20,38 +20,24 @@ import java.math.BigInteger;
import java.util.Arrays;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-class KafkaConsumerIdempotentWithProcessorIT extends KafkaConsumerIdempotentTestSupport {
+public class KafkaConsumerIdempotentWithProcessorIT extends KafkaConsumerIdempotentTestSupport {
public static final String TOPIC = "testidemp3";
+ private int size = 200;
@BindToRegistry("kafkaIdempotentRepository")
private KafkaIdempotentRepository kafkaIdempotentRepository
= new KafkaIdempotentRepository("TEST_IDEMPOTENT", getBootstrapServers());
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerIdempotentWithProcessorIT&autoOffsetReset=earliest"
- + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
- + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("mock:resulti")
- private MockEndpoint to;
-
- private int size = 200;
-
@BeforeEach
public void before() {
- kafkaIdempotentRepository.clear();
kafkaAdminClient.deleteTopics(Arrays.asList(TOPIC, "TEST_IDEMPOTENT")).all();
doSend(size, TOPIC);
}
@@ -64,12 +50,17 @@ class KafkaConsumerIdempotentWithProcessorIT extends KafkaConsumerIdempotentTest
@Override
protected RouteBuilder createRouteBuilder() {
-
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("idemp-with-prop")
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerIdempotentWithProcessorIT&autoOffsetReset=earliest"
+ + "&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ .routeId("idemp-with-prop")
.process(exchange -> {
byte[] id = exchange.getIn().getHeader("id", byte[].class);
@@ -79,13 +70,15 @@ class KafkaConsumerIdempotentWithProcessorIT extends KafkaConsumerIdempotentTest
})
.idempotentConsumer(header("id"))
.idempotentRepository("kafkaIdempotentRepository")
- .to(to);
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@Test
void kafkaMessageIsConsumedByCamel() {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
doRun(to, size);
}
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerLastRecordHeaderIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerLastRecordHeaderIT.java
index 68e098729d5..19b5f950c14 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerLastRecordHeaderIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerLastRecordHeaderIT.java
@@ -20,10 +20,10 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
-import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
@@ -39,9 +39,6 @@ public class KafkaConsumerLastRecordHeaderIT extends BaseEmbeddedKafkaTestSuppor
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerLastRecordHeaderIT.class);
private static final String TOPIC = "last-record";
- @EndpointInject("mock:result")
- private MockEndpoint result;
-
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
@@ -65,6 +62,7 @@ public class KafkaConsumerLastRecordHeaderIT extends BaseEmbeddedKafkaTestSuppor
*/
@Test
public void shouldStartFromBeginningWithEmptyOffsetRepository() throws InterruptedException {
+ MockEndpoint result = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
result.expectedMessageCount(5);
result.expectedBodiesReceived("message-0", "message-1", "message-2", "message-3", "message-4");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerRebalanceIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerRebalanceIT.java
index b56211ff66d..0067981b6d4 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerRebalanceIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerRebalanceIT.java
@@ -21,9 +21,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.spi.StateRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -35,18 +34,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class KafkaConsumerRebalanceIT extends BaseEmbeddedKafkaTestSupport {
private static final String TOPIC = "offset-rebalance";
- @EndpointInject("mock:result")
- private MockEndpoint result;
+ private CountDownLatch messagesLatch = new CountDownLatch(1);
@BindToRegistry("offset")
- private OffsetStateRepository stateRepository;
- private CountDownLatch messagesLatch;
-
- @Override
- protected void doPreSetup() {
- messagesLatch = new CountDownLatch(1);
- stateRepository = new OffsetStateRepository(messagesLatch);
- }
+ private OffsetStateRepository offsetStateRepository = new OffsetStateRepository(messagesLatch);
@Test
public void offsetGetStateMustHaveBeenCalledTwice() throws Exception {
@@ -62,14 +53,14 @@ public class KafkaConsumerRebalanceIT extends BaseEmbeddedKafkaTestSupport {
kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
}
- @Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
from("kafka:" + TOPIC + "?groupId=" + TOPIC + "_GROUP" + "&autoCommitIntervalMs=1000"
+ "&autoOffsetReset=latest" + "&consumersCount=1"
- + "&offsetRepository=#offset").routeId("consumer-rebalance-route").to("mock:result");
+ + "&offsetRepository=#offset").routeId("consumer-rebalance-route")
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerStopIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerStopIT.java
index fe961683dbe..36c93f5b699 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerStopIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerStopIT.java
@@ -21,19 +21,17 @@ import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConsumer;
import org.apache.camel.component.kafka.KafkaFetchRecords;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.platform.commons.function.Try;
import org.junit.platform.commons.util.ReflectionUtils;
@@ -47,22 +45,12 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* This IT is based on {@link KafkaConsumerFullIT}
*/
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class KafkaConsumerStopIT extends BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "test-full";
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerStopIT.class);
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerFullIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
@@ -87,15 +75,21 @@ public class KafkaConsumerStopIT extends BaseEmbeddedKafkaTestSupport {
@Override
public void configure() {
- from(from).process(exchange -> LOG.trace("Captured on the processor: {}",
- exchange.getMessage().getBody()))
- .routeId("full-it").to(to);
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerFullIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ .process(exchange -> LOG.trace("Captured on the processor: {}",
+ exchange.getMessage().getBody()))
+ .routeId("full-it").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@Test
public void kafkaClientConsumerClosedWhenKafkaRouteStopped() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
// given: kafka consumer route
to.expectedBodiesReceivedInAnyOrder("message");
ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", "message");
@@ -106,7 +100,7 @@ public class KafkaConsumerStopIT extends BaseEmbeddedKafkaTestSupport {
org.apache.kafka.clients.consumer.KafkaConsumer kafkaClientConsumer = getKafkaClientConsumer();
// when: kafka consumer route stopped
- context.getRouteController().stopRoute("full-it");
+ contextExtension.getContext().getRouteController().stopRoute("full-it");
// then: org.apache.kafka.clients.consumer.KafkaConsumer closed
await().atMost(10, TimeUnit.SECONDS)
@@ -115,7 +109,7 @@ public class KafkaConsumerStopIT extends BaseEmbeddedKafkaTestSupport {
}
private org.apache.kafka.clients.consumer.KafkaConsumer getKafkaClientConsumer() throws Exception {
- KafkaConsumer kafkaConsumer = (KafkaConsumer) context.getRoute("full-it").getConsumer();
+ KafkaConsumer kafkaConsumer = (KafkaConsumer) contextExtension.getContext().getRoute("full-it").getConsumer();
Try<Object> tasksTry = ReflectionUtils.tryToReadFieldValue(KafkaConsumer.class, "tasks", kafkaConsumer);
KafkaFetchRecords kafkaFetchRecords = ((List<KafkaFetchRecords>) tasksTry.get()).get(0);
Try<Object> kafkaClientConsumerTry
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
index bbae0f74910..c9868ee27fd 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
@@ -20,11 +20,10 @@ import java.util.Collections;
import java.util.Properties;
import java.util.stream.StreamSupport;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
@@ -38,14 +37,6 @@ public class KafkaConsumerTopicIsPatternIT extends BaseEmbeddedKafkaTestSupport
public static final String TOPIC = "vess123d";
public static final String TOPIC_PATTERN = "v.*d";
- @EndpointInject("kafka:" + TOPIC_PATTERN
- + "?topicIsPattern=true&groupId=KafkaConsumerTopicIsPatternIT&autoOffsetReset=earliest"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor&metadataMaxAgeMs=1000")
- private Endpoint from;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
@BeforeEach
@@ -71,13 +62,18 @@ public class KafkaConsumerTopicIsPatternIT extends BaseEmbeddedKafkaTestSupport
@Override
public void configure() {
- from(from).to(to);
+ from("kafka:" + TOPIC_PATTERN
+ + "?topicIsPattern=true&groupId=KafkaConsumerTopicIsPatternIT&autoOffsetReset=earliest"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor&metadataMaxAgeMs=1000")
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@Test
public void kafkaTopicIsPattern() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
to.allMessages().header(KafkaConstants.TOPIC).isEqualTo(TOPIC);
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerFullIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerFullIT.java
index 3906f30ef8c..504ae4e4bef 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerFullIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerFullIT.java
@@ -30,10 +30,7 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.StreamSupport;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
-import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
@@ -51,16 +48,24 @@ import org.apache.kafka.common.header.Headers;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
+import static org.apache.camel.component.kafka.integration.common.TestProducerUtil.sendMessagesInRoute;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
+ public static final String DIRECT_START_STRINGS_URI = "direct:startStrings";
+ public static final String DIRECT_START_STRINGS_2_URI = "direct:startStrings2";
+ public static final String DIRECT_START_BYTES_URI = "direct:startBytes";
+ public static final String DIRECT_START_TRACED_URI = "direct:startTraced";
+ public static final String DIRECT_PROPAGATED_HEADERS_URI = "direct:propagatedHeaders";
+ public static final String DIRECT_NO_RECORD_SPECIFIC_HEADERS_URI = "direct:noRecordSpecificHeaders";
private static final String TOPIC_STRINGS = "test";
private static final String TOPIC_INTERCEPTED = "test";
private static final String TOPIC_STRINGS_IN_HEADER = "testHeader";
@@ -69,57 +74,21 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
private static final String GROUP_BYTES = "groupStrings";
private static final String TOPIC_PROPAGATED_HEADERS = "testPropagatedHeaders";
private static final String TOPIC_NO_RECORD_SPECIFIC_HEADERS = "noRecordSpecificHeaders";
-
+ private static final String KAFKA_ACK_MOCK = "mock:kafkaAck";
private static KafkaConsumer<String, String> stringsConsumerConn;
private static KafkaConsumer<byte[], byte[]> bytesConsumerConn;
-
- @EndpointInject("kafka:" + TOPIC_STRINGS + "?requestRequiredAcks=-1")
- private Endpoint toStrings;
-
- @EndpointInject("kafka:" + TOPIC_STRINGS + "?requestRequiredAcks=-1&partitionKey=0")
- private Endpoint toStrings2;
-
- @EndpointInject("kafka:" + TOPIC_INTERCEPTED + "?requestRequiredAcks=-1"
- + "&interceptorClasses=org.apache.camel.component.kafka.MockProducerInterceptor")
- private Endpoint toStringsWithInterceptor;
-
- @EndpointInject("mock:kafkaAck")
- private MockEndpoint mockEndpoint;
-
- @EndpointInject("kafka:" + TOPIC_BYTES + "?requestRequiredAcks=-1"
- + "&valueSerializer=org.apache.kafka.common.serialization.ByteArraySerializer&"
- + "keySerializer=org.apache.kafka.common.serialization.ByteArraySerializer")
- private Endpoint toBytes;
-
- @EndpointInject("kafka:" + TOPIC_PROPAGATED_HEADERS + "?requestRequiredAcks=-1")
- private Endpoint toPropagatedHeaders;
-
- @EndpointInject("kafka:" + TOPIC_NO_RECORD_SPECIFIC_HEADERS + "?requestRequiredAcks=-1")
- private Endpoint toNoRecordSpecificHeaders;
-
- @Produce("direct:startStrings")
private ProducerTemplate stringsTemplate;
-
- @Produce("direct:startStrings2")
private ProducerTemplate stringsTemplate2;
-
- @Produce("direct:startBytes")
private ProducerTemplate bytesTemplate;
-
- @Produce("direct:startTraced")
private ProducerTemplate interceptedTemplate;
-
- @Produce("direct:propagatedHeaders")
private ProducerTemplate propagatedHeadersTemplate;
-
- @Produce("direct:noRecordSpecificHeaders")
private ProducerTemplate noRecordSpecificHeadersTemplate;
@BindToRegistry("myStrategy")
- private MyHeaderFilterStrategy strategy = new MyHeaderFilterStrategy();
+ private MyHeaderFilterStrategy headerFilterStrategy = new MyHeaderFilterStrategy();
@BindToRegistry("myHeaderSerializer")
- private MyKafkaHeadersSerializer serializer = new MyKafkaHeadersSerializer();
+ private MyKafkaHeadersSerializer headersSerializer = new MyKafkaHeadersSerializer();
@BeforeAll
public static void before() {
@@ -139,22 +108,73 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
kafkaAdminClient.deleteTopics(topics);
}
+ private static KafkaConsumer<String, String> createStringKafkaConsumer(final String groupId) {
+ Properties stringsProps = new Properties();
+
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ return new KafkaConsumer<>(stringsProps);
+ }
+
+ private static KafkaConsumer<byte[], byte[]> createByteKafkaConsumer(final String groupId) {
+ Properties stringsProps = new Properties();
+
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ return new KafkaConsumer<>(stringsProps);
+ }
+
+ @BeforeEach
+ public void setupProducerTemplates() {
+ stringsTemplate = contextExtension.getProducerTemplate();
+ stringsTemplate2 = contextExtension.getProducerTemplate();
+ bytesTemplate = contextExtension.getProducerTemplate();
+ interceptedTemplate = contextExtension.getProducerTemplate();
+ propagatedHeadersTemplate = contextExtension.getProducerTemplate();
+ noRecordSpecificHeadersTemplate = contextExtension.getProducerTemplate();
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from("direct:startStrings").to(toStrings).to(mockEndpoint);
+ from(DIRECT_START_STRINGS_URI).to("kafka:" + TOPIC_STRINGS + "?requestRequiredAcks=-1").to(KAFKA_ACK_MOCK);
- from("direct:startStrings2").to(toStrings2).to(mockEndpoint);
+ from(DIRECT_START_STRINGS_2_URI).to("kafka:" + TOPIC_STRINGS + "?requestRequiredAcks=-1").to(KAFKA_ACK_MOCK);
- from("direct:startBytes").to(toBytes).to(mockEndpoint);
+ from(DIRECT_START_BYTES_URI).to("kafka:" + TOPIC_BYTES + "?requestRequiredAcks=-1"
+ + "&valueSerializer=org.apache.kafka.common.serialization.ByteArraySerializer&"
+ + "keySerializer=org.apache.kafka.common.serialization.ByteArraySerializer")
+ .to(KAFKA_ACK_MOCK);
- from("direct:startTraced").to(toStringsWithInterceptor).to(mockEndpoint);
+ from(DIRECT_START_TRACED_URI).to("kafka:" + TOPIC_INTERCEPTED + "?requestRequiredAcks=-1"
+ + "&interceptorClasses=org.apache.camel.component.kafka.MockProducerInterceptor")
+ .to(KAFKA_ACK_MOCK);
- from("direct:propagatedHeaders").to(toPropagatedHeaders).to(mockEndpoint);
+ from(DIRECT_PROPAGATED_HEADERS_URI).to("kafka:" + TOPIC_PROPAGATED_HEADERS + "?requestRequiredAcks=-1")
+ .to(KAFKA_ACK_MOCK);
- from("direct:noRecordSpecificHeaders").to(toNoRecordSpecificHeaders).to(mockEndpoint);
+ from(DIRECT_NO_RECORD_SPECIFIC_HEADERS_URI)
+ .to("kafka:" + TOPIC_NO_RECORD_SPECIFIC_HEADERS + "?requestRequiredAcks=-1").to(KAFKA_ACK_MOCK);
}
};
}
@@ -166,8 +186,10 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
CountDownLatch messagesLatch = new CountDownLatch(messageInTopic + messageInOtherTopic);
- sendMessagesInRoute(messageInTopic, stringsTemplate, "IT test message", KafkaConstants.PARTITION_KEY, "0");
- sendMessagesInRoute(messageInOtherTopic, stringsTemplate, "IT test message in other topic",
+ ProducerTemplate producerTemplate = contextExtension.getProducerTemplate();
+ sendMessagesInRoute(DIRECT_START_STRINGS_URI, messageInTopic, stringsTemplate, "IT test message",
+ KafkaConstants.PARTITION_KEY, "0");
+ sendMessagesInRoute(DIRECT_START_STRINGS_URI, messageInOtherTopic, stringsTemplate, "IT test message in other topic",
KafkaConstants.PARTITION_KEY, "0", KafkaConstants.TOPIC,
TOPIC_STRINGS_IN_HEADER);
@@ -178,6 +200,7 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
assertTrue(allMessagesReceived,
"Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount());
+ MockEndpoint mockEndpoint = contextExtension.getMockEndpoint(KAFKA_ACK_MOCK);
List<Exchange> exchangeList = mockEndpoint.getExchanges();
assertEquals(15, exchangeList.size(), "Fifteen Exchanges are expected");
for (Exchange exchange : exchangeList) {
@@ -197,8 +220,8 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
CountDownLatch messagesLatch = new CountDownLatch(messageInTopic + messageInOtherTopic);
- sendMessagesInRoute(messageInTopic, stringsTemplate2, "IT test message", (String[]) null);
- sendMessagesInRoute(messageInOtherTopic, stringsTemplate2, "IT test message in other topic",
+ sendMessagesInRoute(DIRECT_START_STRINGS_2_URI, messageInTopic, stringsTemplate2, "IT test message", (String[]) null);
+ sendMessagesInRoute(DIRECT_START_STRINGS_2_URI, messageInOtherTopic, stringsTemplate2, "IT test message in other topic",
KafkaConstants.PARTITION_KEY, "0", KafkaConstants.TOPIC,
TOPIC_STRINGS_IN_HEADER);
@@ -209,6 +232,7 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
assertTrue(allMessagesReceived,
"Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount());
+ MockEndpoint mockEndpoint = contextExtension.getMockEndpoint(KAFKA_ACK_MOCK);
List<Exchange> exchangeList = mockEndpoint.getExchanges();
assertEquals(15, exchangeList.size(), "Fifteen Exchanges are expected");
for (Exchange exchange : exchangeList) {
@@ -228,8 +252,9 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
CountDownLatch messagesLatch = new CountDownLatch(messageInTopic + messageInOtherTopic);
- sendMessagesInRoute(messageInTopic, interceptedTemplate, "IT test message", KafkaConstants.PARTITION_KEY, "0");
- sendMessagesInRoute(messageInOtherTopic, interceptedTemplate, "IT test message in other topic",
+ sendMessagesInRoute(DIRECT_START_TRACED_URI, messageInTopic, interceptedTemplate, "IT test message",
+ KafkaConstants.PARTITION_KEY, "0");
+ sendMessagesInRoute(DIRECT_START_TRACED_URI, messageInOtherTopic, interceptedTemplate, "IT test message in other topic",
KafkaConstants.PARTITION_KEY, "0", KafkaConstants.TOPIC,
TOPIC_STRINGS_IN_HEADER);
createKafkaMessageConsumer(stringsConsumerConn, TOPIC_INTERCEPTED, TOPIC_STRINGS_IN_HEADER, messagesLatch);
@@ -254,12 +279,13 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
msgs.add("Message " + x);
}
- sendMessagesInRoute(1, stringsTemplate, msgs, KafkaConstants.PARTITION_KEY, "0");
+ sendMessagesInRoute(DIRECT_START_STRINGS_URI, 1, stringsTemplate, msgs, KafkaConstants.PARTITION_KEY, "0");
msgs = new ArrayList<>();
for (int x = 0; x < messageInOtherTopic; x++) {
msgs.add("Other Message " + x);
}
- sendMessagesInRoute(1, stringsTemplate, msgs, KafkaConstants.PARTITION_KEY, "0", KafkaConstants.TOPIC,
+ sendMessagesInRoute(DIRECT_START_STRINGS_URI, 1, stringsTemplate, msgs, KafkaConstants.PARTITION_KEY, "0",
+ KafkaConstants.TOPIC,
TOPIC_STRINGS_IN_HEADER);
createKafkaMessageConsumer(stringsConsumerConn, TOPIC_STRINGS, TOPIC_STRINGS_IN_HEADER, messagesLatch);
@@ -268,6 +294,7 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
assertTrue(allMessagesReceived,
"Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount());
+ MockEndpoint mockEndpoint = contextExtension.getMockEndpoint(KAFKA_ACK_MOCK);
List<Exchange> exchangeList = mockEndpoint.getExchanges();
assertEquals(2, exchangeList.size(), "Two Exchanges are expected");
Exchange e1 = exchangeList.get(0);
@@ -297,12 +324,14 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
Map<String, Object> inTopicHeaders = new HashMap<>();
inTopicHeaders.put(KafkaConstants.PARTITION_KEY, "0".getBytes());
- sendMessagesInRoute(messageInTopic, bytesTemplate, "IT test message".getBytes(), inTopicHeaders);
+ sendMessagesInRoute(DIRECT_START_BYTES_URI, messageInTopic, bytesTemplate, "IT test message".getBytes(),
+ inTopicHeaders);
Map<String, Object> otherTopicHeaders = new HashMap<>();
otherTopicHeaders.put(KafkaConstants.PARTITION_KEY, "0".getBytes());
otherTopicHeaders.put(KafkaConstants.TOPIC, TOPIC_BYTES_IN_HEADER);
- sendMessagesInRoute(messageInOtherTopic, bytesTemplate, "IT test message in other topic".getBytes(), otherTopicHeaders);
+ sendMessagesInRoute(DIRECT_START_BYTES_URI, messageInOtherTopic, bytesTemplate,
+ "IT test message in other topic".getBytes(), otherTopicHeaders);
createKafkaBytesMessageConsumer(bytesConsumerConn, TOPIC_BYTES, TOPIC_BYTES_IN_HEADER, messagesLatch);
@@ -311,6 +340,7 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
assertTrue(allMessagesReceived,
"Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount());
+ MockEndpoint mockEndpoint = contextExtension.getMockEndpoint(KAFKA_ACK_MOCK);
List<Exchange> exchangeList = mockEndpoint.getExchanges();
assertEquals(15, exchangeList.size(), "Fifteen Exchanges are expected");
for (Exchange exchange : exchangeList) {
@@ -356,7 +386,7 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
camelHeaders.put("CamelFilteredHeader", "CamelFilteredHeader value");
CountDownLatch messagesLatch = new CountDownLatch(1);
- propagatedHeadersTemplate.sendBodyAndHeaders("Some test message", camelHeaders);
+ propagatedHeadersTemplate.sendBodyAndHeaders(DIRECT_PROPAGATED_HEADERS_URI, "Some test message", camelHeaders);
List<ConsumerRecord<String, String>> records = pollForRecords(createStringKafkaConsumer("propagatedHeaderConsumer"),
TOPIC_PROPAGATED_HEADERS, messagesLatch);
@@ -397,7 +427,8 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
camelHeaders.put(propagatedStringHeaderKey, propagatedStringHeaderValue);
CountDownLatch messagesLatch = new CountDownLatch(1);
- noRecordSpecificHeadersTemplate.sendBodyAndHeaders("Some test message", camelHeaders);
+ noRecordSpecificHeadersTemplate.sendBodyAndHeaders(DIRECT_NO_RECORD_SPECIFIC_HEADERS_URI, "Some test message",
+ camelHeaders);
List<ConsumerRecord<String, String>> records = pollForRecords(
createStringKafkaConsumer("noRecordSpecificHeadersConsumer"), TOPIC_NO_RECORD_SPECIFIC_HEADERS, messagesLatch);
@@ -416,15 +447,16 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
@Test
public void headerFilterStrategyCouldBeOverridden() {
KafkaEndpoint kafkaEndpoint
- = context.getEndpoint("kafka:TOPIC_PROPAGATED_HEADERS?headerFilterStrategy=#myStrategy", KafkaEndpoint.class);
- assertIsInstanceOf(MyHeaderFilterStrategy.class, kafkaEndpoint.getConfiguration().getHeaderFilterStrategy());
+ = contextExtension.getContext().getEndpoint("kafka:TOPIC_PROPAGATED_HEADERS?headerFilterStrategy=#myStrategy",
+ KafkaEndpoint.class);
+ assertInstanceOf(MyHeaderFilterStrategy.class, kafkaEndpoint.getConfiguration().getHeaderFilterStrategy());
}
@Test
public void headerSerializerCouldBeOverridden() {
- KafkaEndpoint kafkaEndpoint = context
+ KafkaEndpoint kafkaEndpoint = contextExtension.getContext()
.getEndpoint("kafka:TOPIC_PROPAGATED_HEADERS?headerSerializer=#myHeaderSerializer", KafkaEndpoint.class);
- assertIsInstanceOf(MyKafkaHeadersSerializer.class, kafkaEndpoint.getConfiguration().getHeaderSerializer());
+ assertInstanceOf(MyKafkaHeadersSerializer.class, kafkaEndpoint.getConfiguration().getHeaderSerializer());
}
private byte[] getHeaderValue(String headerKey, Headers headers) {
@@ -434,40 +466,6 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
return foundHeader.value();
}
- private static KafkaConsumer<String, String> createStringKafkaConsumer(final String groupId) {
- Properties stringsProps = new Properties();
-
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.StringDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.StringDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-
- return new KafkaConsumer<>(stringsProps);
- }
-
- private static KafkaConsumer<byte[], byte[]> createByteKafkaConsumer(final String groupId) {
- Properties stringsProps = new Properties();
-
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootstrapServers());
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.ByteArrayDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.ByteArrayDeserializer");
- stringsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-
- return new KafkaConsumer<>(stringsProps);
- }
-
private List<ConsumerRecord<String, String>> pollForRecords(
KafkaConsumer<String, String> consumerConn, String topic, CountDownLatch messagesLatch) {
@@ -522,22 +520,6 @@ public class KafkaProducerFullIT extends BaseEmbeddedKafkaTestSupport {
}
- private void sendMessagesInRoute(int messages, ProducerTemplate template, Object bodyOther, String... headersWithValue) {
- Map<String, Object> headerMap = new HashMap<>();
- if (headersWithValue != null) {
- for (int i = 0; i < headersWithValue.length; i = i + 2) {
- headerMap.put(headersWithValue[i], headersWithValue[i + 1]);
- }
- }
- sendMessagesInRoute(messages, template, bodyOther, headerMap);
- }
-
- private void sendMessagesInRoute(int messages, ProducerTemplate template, Object bodyOther, Map<String, Object> headerMap) {
- for (int k = 0; k < messages; k++) {
- template.sendBodyAndHeaders(bodyOther, headerMap);
- }
- }
-
private static class MyHeaderFilterStrategy extends DefaultHeaderFilterStrategy {
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaToDIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaToDIT.java
index 8234d5c312d..330a50d1f90 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaToDIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaToDIT.java
@@ -25,11 +25,12 @@ public class KafkaToDIT extends BaseEmbeddedKafkaTestSupport {
@Test
public void testToD() {
- template.sendBodyAndHeader("direct:start", "Hello bar", "where", "bar");
- template.sendBodyAndHeader("direct:start", "Hello beer", "where", "beer");
+ contextExtension.getProducerTemplate().sendBodyAndHeader("direct:start", "Hello bar", "where", "bar");
+ contextExtension.getProducerTemplate().sendBodyAndHeader("direct:start", "Hello beer", "where", "beer");
// there should only be one kafka endpoint
- long count = context.getEndpoints().stream().filter(e -> e.getEndpointUri().startsWith("kafka:")).count();
+ long count = contextExtension.getContext().getEndpoints().stream().filter(e -> e.getEndpointUri().startsWith("kafka:"))
+ .count();
assertEquals(1, count, "There should only be 1 kafka endpoint");
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaTransactionIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaTransactionIT.java
index eefabf142c8..68efdf8088e 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaTransactionIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaTransactionIT.java
@@ -19,21 +19,18 @@ package org.apache.camel.component.kafka.integration;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
+import org.apache.camel.component.kafka.integration.common.TestProducerUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -48,35 +45,13 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
+ public static final String SEQUENTIAL_TRANSACTION_URI = "direct:startTransaction";
+ public static final String CONCURRENT_TRANSACTION_URI = "seda:startTransaction";
+
private static final String TOPIC_TRANSACTION = "transaction";
private static final String TOPIC_CONCURRENCY_TRANSACTION = "concurrency_transaction";
- private static KafkaConsumer<String, String> stringsConsumerConn;
private static final int THREAD_NUM = 5;
-
- @EndpointInject("kafka:" + TOPIC_TRANSACTION + "?requestRequiredAcks=-1"
- + "&additional-properties[transactional.id]=1234"
- + "&additional-properties[enable.idempotence]=true"
- + "&additional-properties[retries]=5")
- private Endpoint toTransaction;
-
- @EndpointInject("kafka:" + TOPIC_CONCURRENCY_TRANSACTION + "?requestRequiredAcks=-1&synchronous=true"
- + "&additional-properties[transactional.id]=5678"
- + "&additional-properties[enable.idempotence]=true"
- + "&additional-properties[retries]=5")
- private Endpoint toConcurrencyTransaction;
-
- @EndpointInject("mock:kafkaAck")
- private MockEndpoint mockEndpoint;
-
- @Produce("direct:startTransaction")
- private ProducerTemplate testTransaction;
-
- @Produce("seda:startTransaction")
- private ProducerTemplate testConcurrencyTransaction;
-
- public KafkaTransactionIT() {
-
- }
+ private static KafkaConsumer<String, String> stringsConsumerConn;
@BeforeAll
public static void before() {
@@ -97,7 +72,10 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:startTransaction").to(toTransaction)
+ from(SEQUENTIAL_TRANSACTION_URI).to("kafka:" + TOPIC_TRANSACTION + "?requestRequiredAcks=-1"
+ + "&additional-properties[transactional.id]=1234"
+ + "&additional-properties[enable.idempotence]=true"
+ + "&additional-properties[retries]=5")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
@@ -106,9 +84,13 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
throw new RuntimeException("fail process message " + body);
}
}
- }).to(mockEndpoint);
+ }).to(KafkaTestUtil.MOCK_RESULT);
- from("seda:startTransaction").to(toConcurrencyTransaction);
+ from(CONCURRENT_TRANSACTION_URI)
+ .to("kafka:" + TOPIC_CONCURRENCY_TRANSACTION + "?requestRequiredAcks=-1&synchronous=true"
+ + "&additional-properties[transactional.id]=5678"
+ + "&additional-properties[enable.idempotence]=true"
+ + "&additional-properties[retries]=5");
}
};
}
@@ -124,7 +106,10 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
threads[i] = new Thread(new Runnable() {
@Override
public void run() {
- sendMessagesInRoute(messageInTopic, testConcurrencyTransaction, "IT test concurrency transaction message",
+ ProducerTemplate testConcurrencyTransaction = contextExtension.getProducerTemplate();
+
+ TestProducerUtil.sendMessagesInRoute(CONCURRENT_TRANSACTION_URI, messageInTopic, testConcurrencyTransaction,
+ "IT test concurrency transaction message",
KafkaConstants.PARTITION_KEY,
"0");
}
@@ -150,11 +135,15 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
CountDownLatch messagesLatch = new CountDownLatch(messageInTopic);
- sendMessagesInRoute(messageInTopic, testTransaction, "IT test transaction message", KafkaConstants.PARTITION_KEY, "0");
+ ProducerTemplate testTransaction = contextExtension.getProducerTemplate();
+ TestProducerUtil.sendMessagesInRoute(SEQUENTIAL_TRANSACTION_URI, messageInTopic, testTransaction,
+ "IT test transaction message",
+ KafkaConstants.PARTITION_KEY, "0");
assertThrows(RuntimeException.class, new Executable() {
@Override
- public void execute() throws Throwable {
- sendMessagesInRoute(messageInTopic, testTransaction, "IT test transaction fail message",
+ public void execute() {
+ TestProducerUtil.sendMessagesInRoute(SEQUENTIAL_TRANSACTION_URI, messageInTopic, testTransaction,
+ "IT test transaction fail message",
KafkaConstants.PARTITION_KEY,
"0");
}
@@ -167,6 +156,8 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
assertTrue(allMessagesReceived,
"Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount());
+ MockEndpoint mockEndpoint = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
List<Exchange> exchangeList = mockEndpoint.getExchanges();
assertEquals(10, exchangeList.size(), "Ten Exchanges are expected");
for (Exchange exchange : exchangeList) {
@@ -214,19 +205,4 @@ public class KafkaTransactionIT extends BaseEmbeddedKafkaTestSupport {
}
}
- private void sendMessagesInRoute(int messages, ProducerTemplate template, Object bodyOther, String... headersWithValue) {
- Map<String, Object> headerMap = new HashMap<>();
- if (headersWithValue != null) {
- for (int i = 0; i < headersWithValue.length; i = i + 2) {
- headerMap.put(headersWithValue[i], headersWithValue[i + 1]);
- }
- }
- sendMessagesInRoute(messages, template, bodyOther, headerMap);
- }
-
- private void sendMessagesInRoute(int messages, ProducerTemplate template, Object bodyOther, Map<String, Object> headerMap) {
- for (int k = 0; k < messages; k++) {
- template.sendBodyAndHeaders(bodyOther, headerMap);
- }
- }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseManualCommitTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/BaseManualCommitTestSupport.java
similarity index 90%
rename from components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseManualCommitTestSupport.java
rename to components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/BaseManualCommitTestSupport.java
index ffbadfc6028..00870088828 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseManualCommitTestSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/BaseManualCommitTestSupport.java
@@ -15,13 +15,14 @@
* limitations under the License.
*/
-package org.apache.camel.component.kafka.integration;
+package org.apache.camel.component.kafka.integration.commit;
import java.util.Collections;
import java.util.Properties;
import org.apache.camel.EndpointInject;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.spi.StateRepository;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -31,7 +32,7 @@ import org.junit.jupiter.api.BeforeEach;
import static org.junit.jupiter.api.Assertions.assertEquals;
-public class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
+abstract class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
@EndpointInject("mock:result")
protected MockEndpoint to;
@@ -65,7 +66,7 @@ public class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
to.reset();
// Second step: We shut down our route, we expect nothing will be recovered by our route
- context.getRouteController().stopRoute("foo");
+ contextExtension.getContext().getRouteController().stopRoute("foo");
to.expectedMessageCount(0);
// Third step: While our route is stopped, we send 3 records more to Kafka test topic
@@ -77,7 +78,7 @@ public class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
// Fourth step: We start again our route, since we have been committing the offsets from the first step,
// we will expect to consume from the latest committed offset e.g from offset 5
- context.getRouteController().startRoute("foo");
+ contextExtension.getContext().getRouteController().startRoute("foo");
setupPostExecutionExpectations();
to.assertIsSatisfied(3000);
@@ -99,7 +100,7 @@ public class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
assertEquals("5", state, "5 messages were sent in the first step, therefore the offset should be 5");
// Second step: We shut down our route, we expect nothing will be recovered by our route
- context.getRouteController().stopRoute("foo");
+ contextExtension.getContext().getRouteController().stopRoute("foo");
to.expectedMessageCount(0);
// Third step: While our route is stopped, we send 3 records more to Kafka test topic
@@ -111,7 +112,7 @@ public class BaseManualCommitTestSupport extends BaseEmbeddedKafkaTestSupport {
// Fourth step: We start again our route, since we have been committing the offsets from the first step,
// we will expect to consume from the latest committed offset e.g from offset 5
- context.getRouteController().startRoute("foo");
+ contextExtension.getContext().getRouteController().startRoute("foo");
setupPostExecutionExpectations();
to.assertIsSatisfied(3000);
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncCommitIT.java
index 5387ba7d3e2..be65f1a3ce9 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncCommitIT.java
@@ -16,12 +16,10 @@
*/
package org.apache.camel.component.kafka.integration.commit;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
-import org.apache.camel.component.kafka.integration.BaseManualCommitTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.RepeatedTest;
@@ -31,11 +29,6 @@ public class KafkaConsumerAsyncCommitIT extends BaseManualCommitTestSupport {
public static final String TOPIC = "testManualAsyncCommitTest";
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerAsyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
- + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory")
- private Endpoint from;
-
@AfterEach
public void after() {
cleanupKafka(TOPIC);
@@ -47,12 +40,19 @@ public class KafkaConsumerAsyncCommitIT extends BaseManualCommitTestSupport {
@Override
public void configure() {
- from(from).routeId("foo").to(to).process(e -> {
- KafkaManualCommit manual = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
- assertNotNull(manual);
- manual.commit();
- });
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerAsyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory")
+ .routeId("foo").to(KafkaTestUtil.MOCK_RESULT).process(e -> {
+ KafkaManualCommit manual
+ = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
+ assertNotNull(manual);
+ manual.commit();
+ });
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerAsyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory")
+ .routeId("bar").autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncWithOffsetRepoCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncWithOffsetRepoCommitIT.java
index dcdc9e990d5..d711fa916f8 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncWithOffsetRepoCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerAsyncWithOffsetRepoCommitIT.java
@@ -17,12 +17,10 @@
package org.apache.camel.component.kafka.integration.commit;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
-import org.apache.camel.component.kafka.integration.BaseManualCommitTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.impl.engine.MemoryStateRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.DisplayName;
@@ -35,18 +33,7 @@ public class KafkaConsumerAsyncWithOffsetRepoCommitIT extends BaseManualCommitTe
public static final String TOPIC = "testAsyncCommitWithOffsetRepoTest";
@BindToRegistry("stateRepository")
- private final MemoryStateRepository stateRepository = new MemoryStateRepository();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerAsyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false&offsetRepository=#bean:stateRepository"
- + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory")
- private Endpoint from;
-
- public static MemoryStateRepository offsetRepo() {
- MemoryStateRepository stateRepository = new MemoryStateRepository();
- stateRepository.setState(TOPIC + "/0", "");
- return stateRepository;
- }
+ private static MemoryStateRepository stateRepository = new MemoryStateRepository();
@AfterEach
public void after() {
@@ -59,15 +46,20 @@ public class KafkaConsumerAsyncWithOffsetRepoCommitIT extends BaseManualCommitTe
@Override
public void configure() {
- from(from)
- .routeId("foo").to(to).process(e -> {
+ final String uri = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerAsyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false&offsetRepository=#bean:stateRepository"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualAsyncCommitFactory";
+
+ from(uri)
+ .routeId("foo").to(KafkaTestUtil.MOCK_RESULT).process(e -> {
KafkaManualCommit manual
= e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
assertNotNull(manual);
manual.commit();
});
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ from(uri)
+ .routeId("bar").autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerNoopCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerNoopCommitIT.java
index 344e5b3f648..e563b21e457 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerNoopCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerNoopCommitIT.java
@@ -16,12 +16,11 @@
*/
package org.apache.camel.component.kafka.integration.commit;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
-import org.apache.camel.component.kafka.integration.BaseManualCommitTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -32,11 +31,6 @@ public class KafkaConsumerNoopCommitIT extends BaseManualCommitTestSupport {
public static final String TOPIC = "testManualNoopCommitTest";
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerNoopCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
- + "&allowManualCommit=true&autoOffsetReset=earliest&metadataMaxAgeMs=1000")
- private Endpoint from;
-
@AfterEach
public void after() {
cleanupKafka(TOPIC);
@@ -48,12 +42,19 @@ public class KafkaConsumerNoopCommitIT extends BaseManualCommitTestSupport {
@Override
public void configure() {
- from(from).routeId("foo").to(to).process(e -> {
- KafkaManualCommit manual = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
- assertNotNull(manual);
- manual.commit();
- });
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerNoopCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&metadataMaxAgeMs=1000").routeId("foo")
+ .to(KafkaTestUtil.MOCK_RESULT).process(e -> {
+ KafkaManualCommit manual
+ = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
+ assertNotNull(manual);
+ manual.commit();
+ });
+ from("kafka:" + TOPIC
+ + "?groupId=KafkaConsumerNoopCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&metadataMaxAgeMs=1000").routeId("bar")
+ .autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
@@ -71,6 +72,7 @@ public class KafkaConsumerNoopCommitIT extends BaseManualCommitTestSupport {
to.reset();
+ CamelContext context = contextExtension.getContext();
context.getRouteController().stopRoute("foo");
to.expectedMessageCount(0);
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncCommitIT.java
index b6eb9f7791d..b145bd20e77 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncCommitIT.java
@@ -16,12 +16,10 @@
*/
package org.apache.camel.component.kafka.integration.commit;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
-import org.apache.camel.component.kafka.integration.BaseManualCommitTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.RepeatedTest;
@@ -31,11 +29,6 @@ public class KafkaConsumerSyncCommitIT extends BaseManualCommitTestSupport {
public static final String TOPIC = "testManualCommitSyncTest";
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerSyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
- + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory")
- private Endpoint from;
-
@AfterEach
public void after() {
cleanupKafka(TOPIC);
@@ -43,16 +36,20 @@ public class KafkaConsumerSyncCommitIT extends BaseManualCommitTestSupport {
@Override
protected RouteBuilder createRouteBuilder() {
+ String from = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerSyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory";
+
return new RouteBuilder() {
@Override
public void configure() {
- from(from).routeId("foo").to(to).process(e -> {
+ from(from).routeId("foo").to(KafkaTestUtil.MOCK_RESULT).process(e -> {
KafkaManualCommit manual = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
assertNotNull(manual);
manual.commit();
});
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ from(from).routeId("bar").autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncWithOffsetRepoCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncWithOffsetRepoCommitIT.java
index 159d1c9499e..e6f02495ed5 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncWithOffsetRepoCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/commit/KafkaConsumerSyncWithOffsetRepoCommitIT.java
@@ -17,12 +17,10 @@
package org.apache.camel.component.kafka.integration.commit;
import org.apache.camel.BindToRegistry;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
-import org.apache.camel.component.kafka.integration.BaseManualCommitTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.impl.engine.MemoryStateRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.DisplayName;
@@ -30,17 +28,12 @@ import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertNotNull;
-public class KafkaConsumerSyncWithOffsetRepoCommitIT extends BaseManualCommitTestSupport {
+class KafkaConsumerSyncWithOffsetRepoCommitIT extends BaseManualCommitTestSupport {
public static final String TOPIC = "testManualCommitSyncWithOffsetRepoTest";
@BindToRegistry("stateRepository")
- private final MemoryStateRepository stateRepository = new MemoryStateRepository();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerSyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false&offsetRepository=#bean:stateRepository"
- + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory")
- private Endpoint from;
+ private static MemoryStateRepository stateRepository = new MemoryStateRepository();
@AfterEach
public void after() {
@@ -53,12 +46,18 @@ public class KafkaConsumerSyncWithOffsetRepoCommitIT extends BaseManualCommitTes
@Override
public void configure() {
- from(from).routeId("foo").to(to).process(e -> {
- KafkaManualCommit manual = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
- assertNotNull(manual);
- manual.commit();
- });
- from(from).routeId("bar").autoStartup(false).to(toBar);
+ final String from = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerSyncCommitIT&pollTimeoutMs=1000&autoCommitEnable=false&offsetRepository=#bean:stateRepository"
+ + "&allowManualCommit=true&autoOffsetReset=earliest&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory";
+ from(from).routeId("foo")
+ .to(KafkaTestUtil.MOCK_RESULT).process(e -> {
+ KafkaManualCommit manual
+ = e.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
+ assertNotNull(manual);
+ manual.commit();
+ });
+ from(from)
+ .routeId("bar").autoStartup(false).to(KafkaTestUtil.MOCK_RESULT_BAR);
}
};
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaAdminUtil.java
similarity index 61%
rename from components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java
rename to components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaAdminUtil.java
index adf2b12068e..62083957a41 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/BaseEmbeddedKafkaAuthTestSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaAdminUtil.java
@@ -14,7 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.kafka.integration;
+
+package org.apache.camel.component.kafka.integration.common;
import java.util.Collections;
import java.util.Map;
@@ -23,80 +24,53 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import org.apache.camel.CamelContext;
import org.apache.camel.test.infra.kafka.services.ContainerLocalAuthKafkaService;
import org.apache.camel.test.infra.kafka.services.KafkaService;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.common.config.SaslConfigs;
import org.junit.Assert;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Tag;
-import org.junit.jupiter.api.Tags;
-import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-@Tags({ @Tag("non-abstract") })
-public abstract class BaseEmbeddedKafkaAuthTestSupport extends AbstractKafkaTestSupport {
- @RegisterExtension
- public static ContainerLocalAuthKafkaService service = new ContainerLocalAuthKafkaService("/kafka-jaas.config");
+public final class KafkaAdminUtil {
- protected static AdminClient kafkaAdminClient;
+ private KafkaAdminUtil() {
- @BeforeAll
- public static void beforeClass() {
- AbstractKafkaTestSupport.setServiceProperties(service);
}
- public static AdminClient createAuthAdminClient(KafkaService service) {
+ public static AdminClient createAdminClient(KafkaService service) {
final Properties properties = new Properties();
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, service.getBootstrapServers());
- properties.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
- properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
- properties.put(SaslConfigs.SASL_JAAS_CONFIG,
- ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("admin", "admin-secret"));
-
- return AdminClient.create(properties);
- }
-
- @BeforeEach
- public void setKafkaAdminClient() {
- if (kafkaAdminClient == null) {
- kafkaAdminClient = createAdminClient();
- }
- }
-
- protected Properties getDefaultProperties() {
- return getDefaultProperties(service);
- }
-
- @Override
- protected CamelContext createCamelContext() throws Exception {
- return createCamelContextFromService(service);
- }
-
- protected static String getBootstrapServers() {
- return service.getBootstrapServers();
- }
- private static AdminClient createAdminClient() {
- return createAuthAdminClient(service);
+ return KafkaAdminClient.create(properties);
}
- protected static Map<String, ConsumerGroupDescription> getConsumerGroupInfo(String groupId)
+ public static Map<String, ConsumerGroupDescription> getConsumerGroupInfo(String groupId, AdminClient kafkaAdminClient)
throws InterruptedException, ExecutionException, TimeoutException {
return kafkaAdminClient.describeConsumerGroups(Collections.singletonList(groupId)).all().get(30, TimeUnit.SECONDS);
}
- protected static void assertGroupIsConnected(String groupId) {
- final Map<String, ConsumerGroupDescription> allGroups = assertDoesNotThrow(() -> getConsumerGroupInfo(groupId));
+ public static void assertGroupIsConnected(String groupId, AdminClient kafkaAdminClient) {
+ final Map<String, ConsumerGroupDescription> allGroups
+ = assertDoesNotThrow(() -> getConsumerGroupInfo(groupId, kafkaAdminClient));
Assert.assertTrue("There should be at least one group named" + groupId, allGroups.size() >= 1);
final ConsumerGroupDescription groupInfo = allGroups.get("KafkaConsumerAuthIT");
Assert.assertNotNull("There should be at least one group named KafkaConsumerAuthIT", groupInfo);
}
+
+ public static AdminClient createAuthAdminClient(KafkaService service) {
+ final Properties properties = new Properties();
+ properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, service.getBootstrapServers());
+ properties.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+ properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+ properties.put(SaslConfigs.SASL_JAAS_CONFIG,
+ ContainerLocalAuthKafkaService.generateSimpleSaslJaasConfig("admin", "admin-secret"));
+
+ return AdminClient.create(properties);
+ }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/AbstractKafkaTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaTestUtil.java
similarity index 72%
rename from components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/AbstractKafkaTestSupport.java
rename to components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaTestUtil.java
index 6bb36ecf4c1..3e85ab5f5d1 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/AbstractKafkaTestSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/KafkaTestUtil.java
@@ -15,15 +15,12 @@
* limitations under the License.
*/
-package org.apache.camel.component.kafka.integration;
+package org.apache.camel.component.kafka.integration.common;
import java.util.Properties;
-import org.apache.camel.CamelContext;
-import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.test.infra.kafka.services.KafkaService;
-import org.apache.camel.test.junit5.CamelTestSupport;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.KafkaAdminClient;
@@ -31,10 +28,18 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public abstract class AbstractKafkaTestSupport extends CamelTestSupport {
- private static final Logger LOG = LoggerFactory.getLogger(AbstractKafkaTestSupport.class);
+public final class KafkaTestUtil {
+ public static final String MOCK_RESULT = "mock:result";
+ public static final String MOCK_RESULT_BAR = "mock:resultBar";
+ public static final String MOCK_DLQ = "mock:dlq";
- protected static void setServiceProperties(KafkaService service) {
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaTestUtil.class);
+
+ private KafkaTestUtil() {
+
+ }
+
+ public static void setServiceProperties(KafkaService service) {
LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
System.setProperty("bootstrapServers", service.getBootstrapServers());
}
@@ -61,19 +66,4 @@ public abstract class AbstractKafkaTestSupport extends CamelTestSupport {
public static Properties getDefaultProperties(KafkaService service) {
return getDefaultProperties(service.getBootstrapServers());
}
-
- protected CamelContext createCamelContextFromService(KafkaService service) throws Exception {
- CamelContext context = super.createCamelContext();
- context.getPropertiesComponent().setLocation("ref:prop");
-
- KafkaComponent kafka = new KafkaComponent(context);
- kafka.init();
- kafka.getConfiguration().setBrokers(service.getBootstrapServers());
- context.addComponent("kafka", kafka);
-
- return context;
- }
-
- protected abstract Properties getDefaultProperties();
-
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/TestProducerUtil.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/TestProducerUtil.java
new file mode 100644
index 00000000000..1f72811f86f
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/common/TestProducerUtil.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.component.kafka.integration.common;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.ProducerTemplate;
+
+public final class TestProducerUtil {
+
+ private TestProducerUtil() {
+
+ }
+
+ public static void sendMessagesInRoute(
+ String uri, int messages, ProducerTemplate template, Object bodyOther, String... headersWithValue) {
+ Map<String, Object> headerMap = new HashMap<>();
+ if (headersWithValue != null) {
+ for (int i = 0; i < headersWithValue.length; i = i + 2) {
+ headerMap.put(headersWithValue[i], headersWithValue[i + 1]);
+ }
+ }
+ sendMessagesInRoute(uri, messages, template, bodyOther, headerMap);
+ }
+
+ public static void sendMessagesInRoute(
+ String uri, int messages, ProducerTemplate template, Object bodyOther, Map<String, Object> headerMap) {
+ for (int k = 0; k < messages; k++) {
+ template.sendBodyAndHeaders(uri, bodyOther, headerMap);
+ }
+ }
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortHealthCheckIT.java
index c0e680cb0e7..d9d03cb79eb 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortHealthCheckIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortHealthCheckIT.java
@@ -17,124 +17,52 @@
package org.apache.camel.component.kafka.integration.health;
import java.util.Collection;
-import java.util.Collections;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
-import org.apache.camel.component.kafka.MockConsumerInterceptor;
-import org.apache.camel.component.kafka.integration.AbstractKafkaTestSupport;
-import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.health.HealthCheck;
import org.apache.camel.health.HealthCheckHelper;
-import org.apache.camel.health.HealthCheckRegistry;
-import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
-import org.apache.camel.test.infra.kafka.services.KafkaService;
-import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.camel.test.infra.core.ContextFixture;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
-import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@DisabledIfSystemProperty(named = "kafka.instance.type", matches = "local-strimzi-container",
disabledReason = "Test infra Kafka runs the Strimzi containers in a way that conflicts with multiple concurrent images")
-public class KafkaConsumerBadPortHealthCheckIT extends CamelTestSupport {
+public class KafkaConsumerBadPortHealthCheckIT extends KafkaHealthCheckTestSupport {
public static final String TOPIC = "test-health";
- @RegisterExtension
- public static KafkaService service = KafkaServiceFactory.createService();
-
- protected static AdminClient kafkaAdminClient;
-
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerBadPortHealthCheckIT.class);
- @BindToRegistry("myHeaderDeserializer")
- private MyKafkaHeaderDeserializer deserializer = new MyKafkaHeaderDeserializer();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerBadPortHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
@EndpointInject("mock:result")
private MockEndpoint to;
- private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
-
- @BeforeEach
- public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
- producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
- MockConsumerInterceptor.recordsCaptured.clear();
- }
-
- @BeforeAll
- public static void beforeClass() {
- LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
- System.setProperty("bootstrapServers", service.getBootstrapServers());
- System.setProperty("brokers", service.getBootstrapServers());
- }
-
- @BeforeEach
- public void setKafkaAdminClient() {
- if (kafkaAdminClient == null) {
- kafkaAdminClient = AbstractKafkaTestSupport.createAdminClient(service);
- }
- }
-
- @AfterEach
- public void after() {
- if (producer != null) {
- producer.close();
- }
- // clean all test topics
- kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
- }
-
@Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext context = super.createCamelContext();
+ @ContextFixture
+ public void configureContext(CamelContext context) {
context.getPropertiesComponent().setLocation("ref:prop");
KafkaComponent kafka = new KafkaComponent(context);
kafka.init();
kafka.getConfiguration().setBrokers(service.getBootstrapServers() + 123);
context.addComponent("kafka", kafka);
-
- // install health check manually (yes a bit cumbersome)
- HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
- registry.setCamelContext(context);
- Object hc = registry.resolveById("context");
- registry.register(hc);
- hc = registry.resolveById("routes");
- registry.register(hc);
- hc = registry.resolveById("consumers");
- registry.register(hc);
- context.setExtension(HealthCheckRegistry.class, registry);
-
- return context;
}
@Override
@@ -143,6 +71,10 @@ public class KafkaConsumerBadPortHealthCheckIT extends CamelTestSupport {
@Override
public void configure() {
+ String from = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerBadPortHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor";
from(from)
.process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
.routeId("test-health-it").to(to);
@@ -152,11 +84,20 @@ public class KafkaConsumerBadPortHealthCheckIT extends CamelTestSupport {
@Order(1)
@Test
- public void kafkaConsumerHealthCheck() throws InterruptedException {
- // health-check liveness should be UP
+ @DisplayName("Tests that liveness reports UP when it's actually up")
+ public void testReportUpWhenIsUp() {
+ CamelContext context = contextExtension.getContext();
+
Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
Assertions.assertTrue(up, "liveness check");
+ }
+
+ @Order(2)
+ @Test
+ @DisplayName("Tests that readiness reports down when it's actually down")
+ public void testReportCorrectlyWhenDown() {
+ CamelContext context = contextExtension.getContext();
// health-check readiness should be down
await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
@@ -167,7 +108,11 @@ public class KafkaConsumerBadPortHealthCheckIT extends CamelTestSupport {
});
Assertions.assertTrue(up2, "readiness check");
});
+ }
+ @Order(3)
+ @Test
+ public void kafkaConsumerHealthCheck() throws InterruptedException {
String propagatedHeaderKey = "PropagatedCustomHeader";
byte[] propagatedHeaderValue = "propagated header value".getBytes();
to.expectedMessageCount(0);
@@ -184,7 +129,4 @@ public class KafkaConsumerBadPortHealthCheckIT extends CamelTestSupport {
to.assertIsSatisfied(3000);
}
-
- private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
- }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortSupervisingHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortSupervisingHealthCheckIT.java
index 052f81674e4..bc804d66b00 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortSupervisingHealthCheckIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerBadPortSupervisingHealthCheckIT.java
@@ -17,107 +17,45 @@
package org.apache.camel.component.kafka.integration.health;
import java.util.Collection;
-import java.util.Collections;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
-import org.apache.camel.component.kafka.MockConsumerInterceptor;
-import org.apache.camel.component.kafka.integration.AbstractKafkaTestSupport;
-import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.health.HealthCheck;
import org.apache.camel.health.HealthCheckHelper;
-import org.apache.camel.health.HealthCheckRegistry;
import org.apache.camel.impl.engine.DefaultSupervisingRouteController;
-import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
import org.apache.camel.spi.SupervisingRouteController;
-import org.apache.camel.test.infra.kafka.services.KafkaService;
-import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.camel.test.infra.core.ContextFixture;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
-import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+import static org.awaitility.Awaitility.await;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@DisabledIfSystemProperty(named = "kafka.instance.type", matches = "local-strimzi-container",
disabledReason = "Test infra Kafka runs the Strimzi containers in a way that conflicts with multiple concurrent images")
-public class KafkaConsumerBadPortSupervisingHealthCheckIT extends CamelTestSupport {
+public class KafkaConsumerBadPortSupervisingHealthCheckIT extends KafkaHealthCheckTestSupport {
public static final String TOPIC = "test-health";
- @RegisterExtension
- public static KafkaService service = KafkaServiceFactory.createService();
-
- protected static AdminClient kafkaAdminClient;
-
private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerBadPortSupervisingHealthCheckIT.class);
- @BindToRegistry("myHeaderDeserializer")
- private MyKafkaHeaderDeserializer deserializer = new MyKafkaHeaderDeserializer();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerBadPortSupervisingHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
-
- @BeforeEach
- public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
- producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
- MockConsumerInterceptor.recordsCaptured.clear();
- }
-
- @BeforeAll
- public static void beforeClass() {
- LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
- System.setProperty("bootstrapServers", service.getBootstrapServers());
- System.setProperty("brokers", service.getBootstrapServers());
- }
-
- @BeforeEach
- public void setKafkaAdminClient() {
- if (kafkaAdminClient == null) {
- kafkaAdminClient = AbstractKafkaTestSupport.createAdminClient(service);
- }
- }
-
- @AfterEach
- public void after() {
- if (producer != null) {
- producer.close();
- }
- // clean all test topics
- kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
- }
-
+ @ContextFixture
@Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext context = super.createCamelContext();
+ public void configureContext(CamelContext context) {
context.getPropertiesComponent().setLocation("ref:prop");
context.setRouteController(new DefaultSupervisingRouteController());
@@ -130,19 +68,6 @@ public class KafkaConsumerBadPortSupervisingHealthCheckIT extends CamelTestSuppo
kafka.init();
kafka.getConfiguration().setBrokers(service.getBootstrapServers() + 123);
context.addComponent("kafka", kafka);
-
- // install health check manually (yes a bit cumbersome)
- HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
- registry.setCamelContext(context);
- Object hc = registry.resolveById("context");
- registry.register(hc);
- hc = registry.resolveById("routes");
- registry.register(hc);
- hc = registry.resolveById("consumers");
- registry.register(hc);
- context.setExtension(HealthCheckRegistry.class, registry);
-
- return context;
}
@Override
@@ -151,33 +76,57 @@ public class KafkaConsumerBadPortSupervisingHealthCheckIT extends CamelTestSuppo
@Override
public void configure() {
- from(from)
+ String uri = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerBadPortSupervisingHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor";
+
+ from(uri)
.process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
- .routeId("test-health-it").to(to);
+ .routeId("test-health-it").to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@Order(1)
@Test
- public void kafkaConsumerHealthCheck() throws InterruptedException {
+ @DisplayName("Tests that liveness reports UP when it's actually up")
+ public void testReportUpWhenIsUp() {
// health-check liveness should be UP
+ CamelContext context = contextExtension.getContext();
+
Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
Assertions.assertTrue(up, "liveness check");
+ }
+
+ @Order(2)
+ @Test
+ @DisplayName("Tests that readiness reports down when it's actually down")
+ public void testReportCorrectlyWhenDown() {
+ CamelContext context = contextExtension.getContext();
// health-check readiness should be down
- await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
- Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
- boolean up2 = res2.stream().allMatch(r -> {
- return r.getState().equals(HealthCheck.State.DOWN) &&
- r.getMessage().stream().allMatch(msg -> msg.contains("port"));
- });
- Assertions.assertTrue(up2, "readiness check");
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> readinessCheck(context));
+ }
+
+ private static void readinessCheck(CamelContext context) {
+ Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
+ boolean up2 = res2.stream().allMatch(r -> {
+ return r.getState().equals(HealthCheck.State.DOWN) &&
+ r.getMessage().stream().allMatch(msg -> msg.contains("port"));
});
+ Assertions.assertTrue(up2, "readiness check");
+ }
+ @Order(3)
+ @Test
+ @DisplayName("I/O test to ensure everything is working as expected")
+ public void kafkaConsumerHealthCheck() throws InterruptedException {
String propagatedHeaderKey = "PropagatedCustomHeader";
byte[] propagatedHeaderValue = "propagated header value".getBytes();
+
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
to.expectedMessageCount(0);
to.expectedMinimumMessageCount(0);
to.expectedNoHeaderReceived();
@@ -192,7 +141,4 @@ public class KafkaConsumerBadPortSupervisingHealthCheckIT extends CamelTestSuppo
to.assertIsSatisfied(3000);
}
-
- private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
- }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerHealthCheckIT.java
new file mode 100644
index 00000000000..636b1e33d96
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerHealthCheckIT.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.integration.health;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.StreamSupport;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.kafka.services.KafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+@Timeout(30)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class KafkaConsumerHealthCheckIT extends KafkaHealthCheckTestSupport {
+ public static final String TOPIC = "test-health";
+ public static final String SKIPPED_HEADER_KEY = "CamelSkippedHeader";
+ public static final String PROPAGATED_CUSTOM_HEADER = "PropagatedCustomHeader";
+ public static final byte[] PROPAGATED_HEADER_VALUE = "propagated header value".getBytes();
+
+ @Order(1)
+ @RegisterExtension
+ public static KafkaService service = KafkaServiceFactory.createService();
+ @Order(2)
+ @RegisterExtension
+ public static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+ protected static AdminClient kafkaAdminClient;
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerHealthCheckIT.class);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() {
+ String from = "kafka:" + TOPIC + "?brokers=" + service.getBootstrapServers()
+ + "&groupId=KafkaConsumerHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor";
+
+ from(from)
+ .process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
+ .routeId("test-health-it").to(KafkaTestUtil.MOCK_RESULT);
+ }
+ };
+ }
+
+ @Override
+ public void configureContext(CamelContext context) {
+ // NO-OP
+ }
+
+ @Order(1)
+ @Test
+ @DisplayName("Tests that liveness reports UP when it's actually up")
+ public void testReportUpWhenIsUp() {
+ // health-check liveness should be UP
+ CamelContext context = contextExtension.getContext();
+ Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
+ boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "liveness check");
+ }
+
+ @Order(2)
+ @Test
+ @DisplayName("Tests that readiness reports UP when it's actually up")
+ public void testReportReadyWhenReady() {
+ CamelContext context = contextExtension.getContext();
+ // health-check readiness should be ready
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
+ boolean up = res2.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "readiness check");
+ });
+ }
+
+ @Order(3)
+ @Test
+ @DisplayName("I/O test to ensure everything is working as expected")
+ public void testIO() throws InterruptedException {
+ to.expectedMessageCount(5);
+ to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
+ to.expectedHeaderValuesReceivedInAnyOrder(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, null, null, null, null, null);
+ to.expectedHeaderReceived(PROPAGATED_CUSTOM_HEADER, PROPAGATED_HEADER_VALUE);
+
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
+ try (KafkaProducer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props)) {
+ for (int k = 0; k < 5; k++) {
+ String msg = "message-" + k;
+ ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
+ data.headers().add(new RecordHeader("CamelSkippedHeader", "skipped header value".getBytes()));
+ data.headers().add(new RecordHeader(PROPAGATED_CUSTOM_HEADER, PROPAGATED_HEADER_VALUE));
+ producer.send(data);
+ }
+ }
+
+ to.assertIsSatisfied(3000);
+ assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
+
+ Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
+ assertFalse(headers.containsKey(SKIPPED_HEADER_KEY), "Should not receive skipped header");
+ assertTrue(headers.containsKey(PROPAGATED_CUSTOM_HEADER), "Should receive propagated header");
+ }
+
+ @Order(4)
+ @Test
+ @DisplayName("Tests that liveness reports UP when it's down")
+ public void testLivenessWhenDown() {
+ CamelContext context = contextExtension.getContext();
+ // and shutdown Kafka which will make readiness report as DOWN
+ service.shutdown();
+
+ // health-check liveness should be UP
+ final Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
+ final boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "liveness check");
+ }
+
+ @Order(5)
+ @Test
+ @DisplayName("Tests that readiness reports down when it's actually down")
+ public void testReadinessWhenDown() {
+ // but health-check readiness should NOT be ready
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(this::readinessCheck);
+ }
+
+ private void readinessCheck() {
+ CamelContext context = contextExtension.getContext();
+
+ Collection<HealthCheck.Result> res2 = HealthCheckHelper.invoke(context);
+ Optional<HealthCheck.Result> down
+ = res2.stream().filter(r -> r.getState().equals(HealthCheck.State.DOWN)).findFirst();
+ Assertions.assertTrue(down.isPresent());
+ String msg = down.get().getMessage().get();
+ Assertions.assertTrue(msg.contains("KafkaConsumer is not ready"));
+ Map<String, Object> map = down.get().getDetails();
+ Assertions.assertEquals(TOPIC, map.get("topic"));
+ Assertions.assertEquals("test-health-it", map.get("route.id"));
+ }
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerUnresolvableHealthCheckIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerUnresolvableHealthCheckIT.java
index 5b6977e15bc..38f6ee284cc 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerUnresolvableHealthCheckIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerUnresolvableHealthCheckIT.java
@@ -17,138 +17,55 @@
package org.apache.camel.component.kafka.integration.health;
import java.util.Collection;
-import java.util.Collections;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
-import org.apache.camel.component.kafka.MockConsumerInterceptor;
-import org.apache.camel.component.kafka.integration.AbstractKafkaTestSupport;
-import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
-import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.health.HealthCheck;
import org.apache.camel.health.HealthCheckHelper;
-import org.apache.camel.health.HealthCheckRegistry;
-import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
-import org.apache.camel.test.infra.kafka.services.KafkaService;
-import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.camel.test.infra.core.ContextFixture;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@DisabledIfSystemProperty(named = "kafka.instance.type", matches = "local-strimzi-container",
disabledReason = "Test infra Kafka runs the Strimzi containers in a way that conflicts with multiple concurrent images")
-public class KafkaConsumerUnresolvableHealthCheckIT extends CamelTestSupport {
+public class KafkaConsumerUnresolvableHealthCheckIT extends KafkaHealthCheckTestSupport {
public static final String TOPIC = "test-health";
- public static KafkaService service = KafkaServiceFactory.createService();
-
- protected static AdminClient kafkaAdminClient;
-
- private static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerUnresolvableHealthCheckIT.class);
-
- @BindToRegistry("myHeaderDeserializer")
- private MyKafkaHeaderDeserializer deserializer = new MyKafkaHeaderDeserializer();
-
- @EndpointInject("kafka:" + TOPIC
- + "?groupId=KafkaConsumerUnresolvableHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
- + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
- private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
-
- @BeforeEach
- public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
- producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
- MockConsumerInterceptor.recordsCaptured.clear();
- }
-
- @BeforeAll
- public static void beforeClass() {
- service.initialize();
-
- LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
- System.setProperty("bootstrapServers", service.getBootstrapServers());
- System.setProperty("brokers", service.getBootstrapServers());
- }
-
- @AfterAll
- public static void afterClass() {
- service.shutdown();
- }
-
- @BeforeEach
- public void setKafkaAdminClient() {
- if (kafkaAdminClient == null) {
- kafkaAdminClient = AbstractKafkaTestSupport.createAdminClient(service);
- }
- }
-
- @AfterEach
- public void after() {
- if (producer != null) {
- producer.close();
- }
- // clean all test topics
- kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC)).all();
- }
-
@Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext context = super.createCamelContext();
+ @ContextFixture
+ public void configureContext(CamelContext context) {
context.getPropertiesComponent().setLocation("ref:prop");
KafkaComponent kafka = new KafkaComponent(context);
kafka.init();
kafka.getConfiguration().setBrokers(service.getBootstrapServers().replace("localhost", "locaIhost"));
context.addComponent("kafka", kafka);
-
- // install health check manually (yes a bit cumbersome)
- HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
- registry.setCamelContext(context);
- Object hc = registry.resolveById("context");
- registry.register(hc);
- hc = registry.resolveById("routes");
- registry.register(hc);
- hc = registry.resolveById("consumers");
- registry.register(hc);
- context.setExtension(HealthCheckRegistry.class, registry);
-
- return context;
}
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
-
@Override
public void configure() {
+ String from = "kafka:" + TOPIC
+ + "?groupId=KafkaConsumerUnresolvableHealthCheckIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ + "valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor";
+
from(from)
.process(exchange -> LOG.trace("Captured on the processor: {}", exchange.getMessage().getBody()))
.routeId("test-health-it").to(to);
@@ -158,22 +75,39 @@ public class KafkaConsumerUnresolvableHealthCheckIT extends CamelTestSupport {
@Order(1)
@Test
- public void kafkaConsumerHealthCheck() throws InterruptedException {
+ @DisplayName("Tests that liveness reports UP when it's actually up")
+ public void testReportUpWhenIsUp() {
// health-check liveness should be UP
+ CamelContext context = contextExtension.getContext();
+
Collection<HealthCheck.Result> res = HealthCheckHelper.invokeLiveness(context);
boolean up = res.stream().allMatch(r -> r.getState().equals(HealthCheck.State.UP));
Assertions.assertTrue(up, "liveness check");
+ }
+
+ @Order(2)
+ @Test
+ @DisplayName("Tests that readiness reports down when it's actually down")
+ public void testReportCorrectlyWhenDown() {
+ CamelContext context = contextExtension.getContext();
// health-check readiness should be down
- await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
- Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
- boolean up2 = res2.stream().allMatch(r -> {
- return r.getState().equals(HealthCheck.State.DOWN) &&
- r.getMessage().stream().allMatch(msg -> msg.contains("bootstrap"));
- });
- Assertions.assertTrue(up2, "readiness check");
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> readinessCheck(context));
+ }
+
+ private static void readinessCheck(CamelContext context) {
+ Collection<HealthCheck.Result> res2 = HealthCheckHelper.invokeReadiness(context);
+ boolean up2 = res2.stream().allMatch(r -> {
+ return r.getState().equals(HealthCheck.State.DOWN) &&
+ r.getMessage().stream().allMatch(msg -> msg.contains("bootstrap"));
});
+ Assertions.assertTrue(up2, "readiness check");
+ }
+ @Order(3)
+ @Test
+ @DisplayName("I/O test to ensure everything is working as expected")
+ public void kafkaConsumerHealthCheck() throws InterruptedException {
String propagatedHeaderKey = "PropagatedCustomHeader";
byte[] propagatedHeaderValue = "propagated header value".getBytes();
to.expectedMessageCount(0);
@@ -190,7 +124,4 @@ public class KafkaConsumerUnresolvableHealthCheckIT extends CamelTestSupport {
to.assertIsSatisfied(3000);
}
-
- private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
- }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaHealthCheckTestSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaHealthCheckTestSupport.java
new file mode 100644
index 00000000000..e03514644c9
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaHealthCheckTestSupport.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.component.kafka.integration.health;
+
+import java.util.Properties;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.component.kafka.MockConsumerInterceptor;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.health.HealthCheckRegistry;
+import org.apache.camel.impl.health.DefaultHealthCheckRegistry;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
+import org.apache.camel.test.infra.core.api.ConfigurableContext;
+import org.apache.camel.test.infra.core.api.ConfigurableRoute;
+import org.apache.camel.test.infra.kafka.services.KafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+abstract class KafkaHealthCheckTestSupport implements ConfigurableRoute, ConfigurableContext {
+ protected static final Logger LOG = LoggerFactory.getLogger(KafkaConsumerUnresolvableHealthCheckIT.class);
+ @Order(1)
+ @RegisterExtension
+ protected static KafkaService service = KafkaServiceFactory.createService();
+
+ @Order(2)
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+ protected org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+ @EndpointInject("mock:result")
+ protected MockEndpoint to;
+
+ @BeforeAll
+ public static void beforeClass() {
+ service.initialize();
+
+ LOG.info("### Embedded Kafka cluster broker list: {}", service.getBootstrapServers());
+ System.setProperty("bootstrapServers", service.getBootstrapServers());
+ System.setProperty("brokers", service.getBootstrapServers());
+ }
+
+ @AfterEach
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ }
+
+ @BeforeEach
+ public void before() {
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ MockConsumerInterceptor.recordsCaptured.clear();
+ }
+
+ @ContextFixture
+ private void configureHealthRegistry(CamelContext context) {
+ // install health check manually (yes a bit cumbersome)
+ HealthCheckRegistry registry = new DefaultHealthCheckRegistry();
+ registry.setCamelContext(context);
+ Object hc = registry.resolveById("context");
+ registry.register(hc);
+ hc = registry.resolveById("routes");
+ registry.register(hc);
+ hc = registry.resolveById("consumers");
+ registry.register(hc);
+ context.setExtension(HealthCheckRegistry.class, registry);
+ }
+
+ @Override
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
+ protected abstract RoutesBuilder createRouteBuilder();
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerCircuitBreakerIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerCircuitBreakerIT.java
index 0bf9911001b..a34869b258c 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerCircuitBreakerIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerCircuitBreakerIT.java
@@ -25,15 +25,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
import org.apache.camel.component.kafka.consumer.errorhandler.KafkaConsumerListener;
-import org.apache.camel.component.kafka.integration.AbstractKafkaTestSupport;
import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -59,46 +57,15 @@ public class KafkaPausableConsumerCircuitBreakerIT extends BaseEmbeddedKafkaTest
private static final Logger LOG = LoggerFactory.getLogger(KafkaPausableConsumerCircuitBreakerIT.class);
private static final int SIMULATED_FAILURES = 5;
- private LongAdder count = new LongAdder();
-
- @EndpointInject("kafka:" + SOURCE_TOPIC
- + "?groupId=KafkaPausableConsumerCircuitBreakerIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
-
- @EndpointInject("direct:intermediate")
- private Endpoint intermediate;
-
- @EndpointInject("mock:result")
- private MockEndpoint to;
-
+ private static LongAdder count = new LongAdder();
+ private static ScheduledExecutorService executorService;
private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
- private ScheduledExecutorService executorService;
-
- @BeforeEach
- public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
- producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
- MockConsumerInterceptor.recordsCaptured.clear();
- }
-
- @AfterEach
- public void after() {
- if (producer != null) {
- producer.close();
- }
- // clean all test topics
- AbstractKafkaTestSupport.createAdminClient(service)
- .deleteTopics(Collections.singletonList(SOURCE_TOPIC)).all();
- }
-
/*
* This is used by pausable to determine whether or not to pause. If returning true, processing continues. If
* returning false, processing pauses.
*/
- private boolean canContinue() {
+ private static boolean canContinue() {
// First one should go through ...
if (count.intValue() <= 1) {
LOG.info("Count is 1, allowing processing to proceed");
@@ -115,14 +82,31 @@ public class KafkaPausableConsumerCircuitBreakerIT extends BaseEmbeddedKafkaTest
return false;
}
- public void increment() {
+ public static void increment() {
count.increment();
}
- public int getCount() {
+ public static int getCount() {
return count.intValue();
}
+ @BeforeEach
+ public void before() {
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ MockConsumerInterceptor.recordsCaptured.clear();
+ }
+
+ @AfterEach
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ // clean all test topics
+ KafkaTestUtil.createAdminClient(service)
+ .deleteTopics(Collections.singletonList(SOURCE_TOPIC)).all();
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -159,15 +143,18 @@ public class KafkaPausableConsumerCircuitBreakerIT extends BaseEmbeddedKafkaTest
// Binds the configuration to the registry
getCamelContext().getRegistry().bind("pausableCircuit", circuitBreaker);
- from(from)
+ from("kafka:" + SOURCE_TOPIC
+ + "?groupId=KafkaPausableConsumerCircuitBreakerIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
.pausable(new KafkaConsumerListener(), o -> canContinue())
.routeId("pausable-it")
.process(exchange -> LOG.info("Got record from Kafka: {}", exchange.getMessage().getBody()))
.circuitBreaker()
.resilience4jConfiguration().circuitBreaker("pausableCircuit").end()
- .to(intermediate);
+ .to("direct:intermediate");
- from(intermediate)
+ from("direct:intermediate")
.process(exchange -> {
LOG.info("Got record on the intermediate processor: {}", exchange.getMessage().getBody());
@@ -175,7 +162,7 @@ public class KafkaPausableConsumerCircuitBreakerIT extends BaseEmbeddedKafkaTest
throw new RuntimeCamelException("Error");
}
})
- .to(to)
+ .to(KafkaTestUtil.MOCK_RESULT)
.end();
}
};
@@ -190,6 +177,8 @@ public class KafkaPausableConsumerCircuitBreakerIT extends BaseEmbeddedKafkaTest
// Although all messages will be sent more than once to the exception only 5 messages should reach the final
// destination, because sending them on the first few tries should fail
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerIT.java
index 4a6d30debc8..1cc0e8d5952 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/pause/KafkaPausableConsumerIT.java
@@ -24,16 +24,14 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointInject;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
import org.apache.camel.component.kafka.consumer.errorhandler.KafkaConsumerListener;
import org.apache.camel.component.kafka.consumer.support.ProcessingResult;
-import org.apache.camel.component.kafka.integration.AbstractKafkaTestSupport;
import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
+import org.apache.camel.component.kafka.integration.common.KafkaTestUtil;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -54,52 +52,35 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class KafkaPausableConsumerIT extends BaseEmbeddedKafkaTestSupport {
- // Just a wrapper for us to check if the expected methods are being called
- private static class TestListener extends KafkaConsumerListener {
- volatile boolean afterConsumeCalled;
- volatile boolean afterProcessCalled;
-
- @Override
- public boolean afterConsume(Object ignored) {
- afterConsumeCalled = true;
- return super.afterConsume(ignored);
- }
-
- @Override
- public boolean afterProcess(ProcessingResult result) {
- afterProcessCalled = true;
- return super.afterProcess(result);
- }
- }
-
public static final String SOURCE_TOPIC = "pause-source";
-
private static final Logger LOG = LoggerFactory.getLogger(KafkaPausableConsumerIT.class);
-
private static final int RETRY_COUNT = 10;
- private LongAdder count = new LongAdder();
-
- @EndpointInject("kafka:" + SOURCE_TOPIC
- + "?groupId=KafkaPausableConsumerIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
- + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
- private Endpoint from;
+ private static LongAdder count = new LongAdder();
+ private static TestListener testConsumerListener = new TestListener();
+ private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
- @EndpointInject("direct:intermediate")
- private Endpoint intermediate;
+ private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
- @EndpointInject("mock:result")
- private MockEndpoint to;
+ private static boolean canContinue() {
+ // First one should go through ...
+ if (count.intValue() <= 1) {
+ return true;
+ }
- private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+ if (count.intValue() >= RETRY_COUNT) {
+ return true;
+ }
- private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+ return false;
+ }
- private TestListener testConsumerListener = new TestListener();
+ public static int getCount() {
+ return count.intValue();
+ }
@BeforeEach
public void before() {
- Properties props = AbstractKafkaTestSupport.getDefaultProperties(service);
+ Properties props = KafkaTestUtil.getDefaultProperties(service);
producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
MockConsumerInterceptor.recordsCaptured.clear();
@@ -112,46 +93,32 @@ public class KafkaPausableConsumerIT extends BaseEmbeddedKafkaTestSupport {
producer.close();
}
// clean all test topics
- AbstractKafkaTestSupport.createAdminClient(service)
+ KafkaTestUtil.createAdminClient(service)
.deleteTopics(Collections.singletonList(SOURCE_TOPIC)).all();
executorService.shutdownNow();
}
- private boolean canContinue() {
- // First one should go through ...
- if (count.intValue() <= 1) {
- return true;
- }
-
- if (count.intValue() >= RETRY_COUNT) {
- return true;
- }
-
- return false;
- }
-
public void increment() {
count.increment();
}
- public int getCount() {
- return count.intValue();
- }
-
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from(from)
+ from("kafka:" + SOURCE_TOPIC
+ + "?groupId=KafkaPausableConsumerIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ + "&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
.pausable(testConsumerListener, o -> canContinue())
.routeId("pausable-it")
.process(exchange -> LOG.info("Got record from Kafka: {}", exchange.getMessage().getBody()))
- .to(intermediate);
+ .to("direct:intermediate");
- from(intermediate)
+ from("direct:intermediate")
.process(exchange -> {
LOG.info("Got record on the intermediate processor: {}", exchange.getMessage().getBody());
@@ -159,7 +126,7 @@ public class KafkaPausableConsumerIT extends BaseEmbeddedKafkaTestSupport {
throw new RuntimeCamelException("Error");
}
})
- .to(to);
+ .to(KafkaTestUtil.MOCK_RESULT);
}
};
}
@@ -173,6 +140,8 @@ public class KafkaPausableConsumerIT extends BaseEmbeddedKafkaTestSupport {
// Although all messages will be sent more than once to the exception only 5 messages should reach the final
// destination, because sending them on the first few tries should fail
+
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
@@ -205,4 +174,22 @@ public class KafkaPausableConsumerIT extends BaseEmbeddedKafkaTestSupport {
assertFalse(headers.containsKey(skippedHeaderKey), "Should not receive skipped header");
assertTrue(headers.containsKey(propagatedHeaderKey), "Should receive propagated header");
}
+
+ // Just a wrapper for us to check if the expected methods are being called
+ private static class TestListener extends KafkaConsumerListener {
+ volatile boolean afterConsumeCalled;
+ volatile boolean afterProcessCalled;
+
+ @Override
+ public boolean afterConsume(Object ignored) {
+ afterConsumeCalled = true;
+ return super.afterConsume(ignored);
+ }
+
+ @Override
+ public boolean afterProcess(ProcessingResult result) {
+ afterProcessCalled = true;
+ return super.afterProcess(result);
+ }
+ }
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryEagerIT.java b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryEagerIT.java
index e95a0f0642e..fd2584c417f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryEagerIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryEagerIT.java
@@ -18,36 +18,27 @@ package org.apache.camel.processor.idempotent.kafka;
import java.util.UUID;
+import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelExecutionException;
-import org.apache.camel.EndpointInject;
-import org.apache.camel.RoutesBuilder;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* Test for eager idempotentRepository usage.
*/
-@TestInstance(TestInstance.Lifecycle.PER_METHOD)
-public class KafkaIdempotentRepositoryEagerIT extends BaseEmbeddedKafkaTestSupport {
+public class KafkaIdempotentRepositoryEagerIT extends SimpleIdempotentTest {
- private KafkaIdempotentRepository kafkaIdempotentRepository;
-
- @EndpointInject("mock:out")
- private MockEndpoint mockOut;
-
- @EndpointInject("mock:before")
- private MockEndpoint mockBefore;
+ @BindToRegistry("kafkaIdempotentRepositoryEager")
+ private KafkaIdempotentRepository idempotentRepository
+ = new KafkaIdempotentRepository("TEST_EAGER_" + UUID.randomUUID(), service.getBootstrapServers());
@Override
- protected RoutesBuilder createRouteBuilder() {
+ protected RouteBuilder createRouteBuilder() {
// Every instance of the repository must use a different topic to guarantee isolation between tests
- kafkaIdempotentRepository = new KafkaIdempotentRepository("TEST_EAGER_" + UUID.randomUUID(), getBootstrapServers());
- context.getRegistry().bind("kafkaIdempotentRepositoryEager", kafkaIdempotentRepository);
return new RouteBuilder() {
@Override
@@ -60,16 +51,22 @@ public class KafkaIdempotentRepositoryEagerIT extends BaseEmbeddedKafkaTestSuppo
@Test
public void testRemovesDuplicates() {
+ ProducerTemplate template = contextExtension.getProducerTemplate();
+
for (int i = 0; i < 10; i++) {
template.sendBodyAndHeader("direct:in", "Test message", "id", i % 5);
}
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
assertEquals(5, mockOut.getReceivedCounter());
+
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
assertEquals(10, mockBefore.getReceivedCounter());
}
@Test
public void testRollsBackOnException() {
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
mockOut.whenAnyExchangeReceived(exchange -> {
int id = exchange.getIn().getHeader("id", Integer.class);
if (id == 0) {
@@ -77,6 +74,7 @@ public class KafkaIdempotentRepositoryEagerIT extends BaseEmbeddedKafkaTestSuppo
}
});
+ ProducerTemplate template = contextExtension.getProducerTemplate();
for (int i = 0; i < 10; i++) {
try {
template.sendBodyAndHeader("direct:in", "Test message", "id", i % 5);
@@ -85,10 +83,11 @@ public class KafkaIdempotentRepositoryEagerIT extends BaseEmbeddedKafkaTestSuppo
}
}
- assertEquals(6, mockOut.getReceivedCounter()); // id{0} goes through the
- // idempotency check
- // twice
- assertEquals(10, mockBefore.getReceivedCounter());
+ assertEquals(5, mockOut.getReceivedCounter(),
+ "Only the 5 messages from the previous test should have been received ");
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
+ assertEquals(20, mockBefore.getReceivedCounter(),
+ "Test should have received 20 messages in total from all the tests");
}
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryNonEagerIT.java b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryNonEagerIT.java
index 4df97548cdf..cb5b5da2757 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryNonEagerIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryNonEagerIT.java
@@ -19,14 +19,19 @@ package org.apache.camel.processor.idempotent.kafka;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.CamelContext;
import org.apache.camel.CamelExecutionException;
-import org.apache.camel.EndpointInject;
-import org.apache.camel.RoutesBuilder;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
+import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestMethodOrder;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -34,46 +39,57 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* Test for non-eager idempotentRepository usage.
*/
-@TestInstance(TestInstance.Lifecycle.PER_METHOD)
-public class KafkaIdempotentRepositoryNonEagerIT extends BaseEmbeddedKafkaTestSupport {
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class KafkaIdempotentRepositoryNonEagerIT extends SimpleIdempotentTest {
- private KafkaIdempotentRepository kafkaIdempotentRepository;
+ @BindToRegistry("kafkaIdempotentRepositoryNonEager")
+ private KafkaIdempotentRepository kafkaIdempotentRepository
+ = new KafkaIdempotentRepository("TEST_NON_EAGER_" + UUID.randomUUID(), service.getBootstrapServers());
- @EndpointInject("mock:out")
- private MockEndpoint mockOut;
+ @ContextFixture
+ public void configureKafka(CamelContext context) {
+ context.getPropertiesComponent().setLocation("ref:prop");
- @EndpointInject("mock:before")
- private MockEndpoint mockBefore;
+ KafkaComponent kafka = new KafkaComponent(context);
+ kafka.init();
+ kafka.getConfiguration().setBrokers(service.getBootstrapServers());
+ context.addComponent("kafka", kafka);
+ }
@Override
- protected RoutesBuilder createRouteBuilder() {
- // Every instance of the repository must use a different topic to guarantee isolation between tests
- kafkaIdempotentRepository = new KafkaIdempotentRepository("TEST_NON_EAGER_" + UUID.randomUUID(), getBootstrapServers());
- context.getRegistry().bind("kafkaIdempotentRepositoryNonEager", kafkaIdempotentRepository);
-
+ protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from("direct:in").to("mock:before").idempotentConsumer(header("id"))
- .idempotentRepository("kafkaIdempotentRepositoryNonEager").eager(false).to("mock:out").end();
+ from("direct:in").to("mock:before")
+ .idempotentConsumer(header("id")).idempotentRepository("kafkaIdempotentRepositoryNonEager").eager(false)
+ .to("mock:out").end();
}
};
}
+ @Order(1)
@Test
+ @DisplayName("Tests that duplicated messages do not go through")
public void testRemovesDuplicates() {
+ ProducerTemplate template = contextExtension.getProducerTemplate();
for (int i = 0; i < 10; i++) {
template.sendBodyAndHeader("direct:in", "Test message", "id", i % 5);
}
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
await().atMost(5, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(5, mockOut.getReceivedCounter()));
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
assertEquals(10, mockBefore.getReceivedCounter());
}
+ @Order(2)
@Test
+ @DisplayName("Tests that processing exceptions cause the message to be rolled back")
public void testRollsBackOnException() {
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
mockOut.whenAnyExchangeReceived(exchange -> {
int id = exchange.getIn().getHeader("id", Integer.class);
if (id == 0) {
@@ -81,6 +97,7 @@ public class KafkaIdempotentRepositoryNonEagerIT extends BaseEmbeddedKafkaTestSu
}
});
+ ProducerTemplate template = contextExtension.getProducerTemplate();
for (int i = 0; i < 10; i++) {
try {
template.sendBodyAndHeader("direct:in", "Test message", "id", i % 5);
@@ -89,10 +106,12 @@ public class KafkaIdempotentRepositoryNonEagerIT extends BaseEmbeddedKafkaTestSu
}
}
- assertEquals(6, mockOut.getReceivedCounter()); // id{0} goes through the
- // idempotency check
- // twice
- assertEquals(10, mockBefore.getReceivedCounter());
+ assertEquals(5, mockOut.getReceivedCounter(),
+ "Only the 5 messages from the previous test should have been received ");
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
+
+ assertEquals(20, mockBefore.getReceivedCounter(),
+ "Test should have received 20 messages in total from all the tests");
}
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryPersistenceIT.java b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryPersistenceIT.java
index 4ad1a0f57b7..0c76cedf03c 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryPersistenceIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryPersistenceIT.java
@@ -21,11 +21,13 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
-import org.apache.camel.EndpointInject;
-import org.apache.camel.RoutesBuilder;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.integration.BaseEmbeddedKafkaTestSupport;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.api.ConfigurableContext;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
@@ -50,26 +52,24 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
* annotations.
*/
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
-@TestInstance(TestInstance.Lifecycle.PER_METHOD)
-public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTestSupport {
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTestSupport implements ConfigurableContext {
private KafkaIdempotentRepository kafkaIdempotentRepository;
- @EndpointInject("mock:out")
- private MockEndpoint mockOut;
-
- @EndpointInject("mock:before")
- private MockEndpoint mockBefore;
-
void clearTopics() {
kafkaAdminClient.deleteTopics(Arrays.asList("TEST_PERSISTENCE")).all();
}
@Override
- protected RoutesBuilder createRouteBuilder() {
+ @ContextFixture
+ public void configureContext(CamelContext context) {
kafkaIdempotentRepository = new KafkaIdempotentRepository("TEST_PERSISTENCE", getBootstrapServers());
context.getRegistry().bind("kafkaIdempotentRepositoryPersistence", kafkaIdempotentRepository);
+ }
+ @Override
+ protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
@@ -80,6 +80,8 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
}
private void sendMessages(long count) {
+ ProducerTemplate template = contextExtension.getProducerTemplate();
+
for (int i = 0; i < count; i++) {
template.sendBodyAndHeader("direct:in", "Test message", "id", i % 5);
}
@@ -94,10 +96,12 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
sendMessages(count);
// all records sent initially
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
await().atMost(10, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(count, mockBefore.getReceivedCounter()));
// only first 5 records are received, the rest are filtered
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
assertEquals(5, mockOut.getReceivedCounter());
}
@@ -111,10 +115,12 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
sendMessages(count);
// all records sent initially
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
await().atMost(10, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(count, mockBefore.getReceivedCounter()));
// nothing pass the idempotent consumer this time
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
assertEquals(0, mockOut.getReceivedCounter());
}
@@ -130,10 +136,12 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
}
// all records sent initially
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
await().atMost(10, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(count * passes, mockBefore.getReceivedCounter()));
// nothing gets passed the idempotent consumer this time
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
assertEquals(0, mockOut.getReceivedCounter());
}
@@ -149,16 +157,19 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
@Test
@DisplayName("Checks that the remaining messages can finally go through")
public void testFourthPass() {
+ ProducerTemplate template = contextExtension.getProducerTemplate();
int count = 5;
for (int i = 5; i < 10; i++) {
template.sendBodyAndHeader("direct:in", "Test message", "id", i);
}
// all records sent initially
+ MockEndpoint mockBefore = contextExtension.getMockEndpoint("mock:before");
await().atMost(10, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(count, mockBefore.getReceivedCounter()));
// there are no duplicate messages on this run so all of them should pass
+ MockEndpoint mockOut = contextExtension.getMockEndpoint("mock:out");
assertEquals(count, mockOut.getReceivedCounter());
}
@@ -171,5 +182,4 @@ public class KafkaIdempotentRepositoryPersistenceIT extends BaseEmbeddedKafkaTes
clearTopics();
}
-
}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/SimpleIdempotentTest.java b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/SimpleIdempotentTest.java
new file mode 100644
index 00000000000..265c62cdd80
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/SimpleIdempotentTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.processor.idempotent.kafka;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultContextLifeCycleManager;
+import org.apache.camel.test.infra.core.RouteFixture;
+import org.apache.camel.test.infra.kafka.services.KafkaService;
+import org.apache.camel.test.infra.kafka.services.KafkaServiceFactory;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public abstract class SimpleIdempotentTest {
+ @Order(1)
+ @RegisterExtension
+ protected static KafkaService service = KafkaServiceFactory.createSingletonService();
+
+ @Order(2)
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension(
+ new DefaultContextLifeCycleManager(DefaultContextLifeCycleManager.DEFAULT_SHUTDOWN_TIMEOUT, false));
+
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
+ protected abstract RouteBuilder createRouteBuilder();
+}
[camel] 02/04: CAMEL-18957: convert camel-amqp to the new camel-test-infra-core
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6537f0fd563e60ac8b5f2b71a9a5b73c85586939
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Thu Jan 19 11:54:15 2023 +0100
CAMEL-18957: convert camel-amqp to the new camel-test-infra-core
---
components/camel-amqp/pom.xml | 8 ++++
.../apache/camel/component/amqp/AMQPRouteTest.java | 45 +++++++++++-------
.../component/amqp/AMQPRouteTraceFrameTest.java | 53 +++++++++++++++-------
.../camel/component/amqp/AMQPTestSupport.java | 19 ++++----
.../component/amqp/AMQPToDSendDynamicTest.java | 38 +++++++++++++---
.../apache/camel/component/amqp/AMQPToDTest.java | 38 ++++++++++++----
.../amqp/artemis/AMQPEmbeddedBrokerTest.java | 51 ++++++++++++---------
7 files changed, 172 insertions(+), 80 deletions(-)
diff --git a/components/camel-amqp/pom.xml b/components/camel-amqp/pom.xml
index dbc80840c5b..141cbd5d5c3 100644
--- a/components/camel-amqp/pom.xml
+++ b/components/camel-amqp/pom.xml
@@ -75,6 +75,14 @@
<!-- testing -->
<!-- test infra -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test-infra-artemis</artifactId>
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTest.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTest.java
index 9ccb3c7dbdf..0e8281eabff 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTest.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTest.java
@@ -26,12 +26,15 @@ import jakarta.jms.Session;
import jakarta.jms.TextMessage;
import org.apache.camel.CamelContext;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.qpid.jms.message.JmsMessage;
import org.apache.qpid.jms.provider.amqp.message.AmqpJmsMessageFacade;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.camel.component.amqp.AMQPComponent.amqpComponent;
@@ -40,15 +43,22 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class AMQPRouteTest extends AMQPTestSupport {
- @EndpointInject("mock:result")
- MockEndpoint resultEndpoint;
+ private ProducerTemplate template;
- String expectedBody = "Hello there!";
+ private MockEndpoint resultEndpoint;
+ private String expectedBody = "Hello there!";
+
+ @BeforeEach
+ void setupTemplate() {
+ template = contextExtension.getProducerTemplate();
+ resultEndpoint = contextExtension.getMockEndpoint("mock:result");
+ }
@Test
public void testJmsQueue() throws Exception {
resultEndpoint.expectedMessageCount(1);
resultEndpoint.message(0).header("cheese").isEqualTo(123);
+
template.sendBodyAndHeader("amqp-customized:queue:ping", expectedBody, "cheese", 123);
resultEndpoint.assertIsSatisfied();
}
@@ -88,7 +98,7 @@ public class AMQPRouteTest extends AMQPTestSupport {
resultEndpoint.message(0).header("cheese").isEqualTo(123);
// default doesn't map annotations to headers
resultEndpoint.message(0).header("JMS_AMQP_MA_cheese").isNull();
- sendAmqpMessage(context.getComponent("amqp-customized", AMQPComponent.class),
+ sendAmqpMessage(contextExtension.getContext().getComponent("amqp-customized", AMQPComponent.class),
"ping", expectedBody, facade -> {
try {
facade.setApplicationProperty("cheese", 123);
@@ -105,7 +115,7 @@ public class AMQPRouteTest extends AMQPTestSupport {
resultEndpoint.expectedMessageCount(1);
resultEndpoint.message(0).header("cheese").isEqualTo(123);
resultEndpoint.message(0).header("JMS_AMQP_MA_cheese").isEqualTo(456);
- sendAmqpMessage(context.getComponent("amqp-customized2", AMQPComponent.class),
+ sendAmqpMessage(contextExtension.getContext().getComponent("amqp-customized2", AMQPComponent.class),
"ping2", expectedBody, facade -> {
try {
facade.setApplicationProperty("cheese", 123);
@@ -131,21 +141,22 @@ public class AMQPRouteTest extends AMQPTestSupport {
}
}
- // Routes fixtures
+ @ContextFixture
+ public void configureContext(CamelContext context) {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext camelContext = super.createCamelContext();
- camelContext.getRegistry().bind("amqpConnection", discoverAMQP(camelContext));
+ context.getRegistry().bind("amqpConnection", discoverAMQP(context));
+ context.addComponent("amqp-customized", amqpComponent(service.serviceAddress()));
+ context.addComponent("amqp-customized2", amqpComponent(service.serviceAddress()));
+ context.getComponent("amqp-customized2", AMQPComponent.class).setIncludeAmqpAnnotations(true);
+ }
- camelContext.addComponent("amqp-customized", amqpComponent(service.serviceAddress()));
- camelContext.addComponent("amqp-customized2", amqpComponent(service.serviceAddress()));
- camelContext.getComponent("amqp-customized2", AMQPComponent.class).setIncludeAmqpAnnotations(true);
- return camelContext;
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
}
- @Override
- protected RouteBuilder createRouteBuilder() {
+ private static RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
from("amqp-customized:queue:ping")
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTraceFrameTest.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTraceFrameTest.java
index 601ca9a0680..f4c63d6d01f 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTraceFrameTest.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPRouteTraceFrameTest.java
@@ -17,35 +17,55 @@
package org.apache.camel.component.amqp;
import org.apache.camel.CamelContext;
-import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.artemis.services.ArtemisService;
+import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.apache.qpid.jms.JmsConnectionFactory;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.apache.camel.component.amqp.AMQPComponent.amqpComponent;
import static org.apache.camel.component.amqp.AMQPConnectionDetails.discoverAMQP;
public class AMQPRouteTraceFrameTest extends AMQPTestSupport {
- @EndpointInject("mock:result")
- MockEndpoint resultEndpoint;
+ @RegisterExtension
+ protected static ArtemisService service = ArtemisServiceFactory.createSingletonAMQPService();
- String expectedBody = "Hello there!";
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
+
+ private MockEndpoint resultEndpoint;
+
+ private String expectedBody = "Hello there!";
+ private ProducerTemplate template;
+
+ @BeforeEach
+ void setupTemplate() {
+ resultEndpoint = contextExtension.getMockEndpoint("mock:result");
+ template = contextExtension.getProducerTemplate();
+ }
@Test
public void testTraceFrame() throws Exception {
resultEndpoint.expectedMessageCount(1);
resultEndpoint.message(0).header("cheese").isEqualTo(123);
- template.sendBodyAndHeader("amqp-customized:queue:ping", expectedBody, "cheese", 123);
+
+ template.sendBodyAndHeader("amqp-with-trace:queue:ping", expectedBody, "cheese", 123);
resultEndpoint.assertIsSatisfied();
}
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext camelContext = super.createCamelContext();
-
- camelContext.getRegistry().bind("amqpConnection", discoverAMQP(camelContext));
+ @ContextFixture
+ public void configureContext(CamelContext context) {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
+ context.getRegistry().bind("amqpConnection", discoverAMQP(context));
JmsConnectionFactory connectionFactory
= new JmsConnectionFactory(service.serviceAddress() + "?amqp.traceFrames=true");
@@ -53,18 +73,17 @@ public class AMQPRouteTraceFrameTest extends AMQPTestSupport {
AMQPComponent amqp = amqpComponent(service.serviceAddress());
amqp.getConfiguration().setConnectionFactory(connectionFactory);
- camelContext.addComponent("amqp-customized", amqp);
- return camelContext;
+ context.addComponent("amqp-with-trace", amqp);
}
- @Override
- protected RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(new RouteBuilder() {
public void configure() {
- from("amqp-customized:queue:ping")
+ from("amqp-with-trace:queue:ping")
.to("log:routing")
.to("mock:result");
}
- };
+ });
}
}
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java
index 266cecf99a0..60790fcd0fb 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPTestSupport.java
@@ -14,21 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.camel.component.amqp;
import org.apache.camel.test.infra.artemis.services.ArtemisService;
import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeAll;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.api.ConfigurableContext;
+import org.apache.camel.test.infra.core.api.ConfigurableRoute;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.extension.RegisterExtension;
-public class AMQPTestSupport extends CamelTestSupport {
-
+public abstract class AMQPTestSupport implements ConfigurableContext, ConfigurableRoute {
+ @Order(1)
@RegisterExtension
protected static ArtemisService service = ArtemisServiceFactory.createSingletonAMQPService();
- @BeforeAll
- public static void beforeAll() {
- System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
- }
+ @Order(2)
+ @RegisterExtension
+ protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
}
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDSendDynamicTest.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDSendDynamicTest.java
index 545da3dc360..96a987deadd 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDSendDynamicTest.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDSendDynamicTest.java
@@ -17,13 +17,21 @@
package org.apache.camel.component.amqp;
import org.apache.camel.CamelContext;
+import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.RouteFixture;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.camel.component.amqp.AMQPConnectionDetails.discoverAMQP;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class AMQPToDSendDynamicTest extends AMQPTestSupport {
+ private ProducerTemplate template;
+ private ConsumerTemplate consumer;
@Test
public void testToD() {
@@ -31,7 +39,8 @@ public class AMQPToDSendDynamicTest extends AMQPTestSupport {
template.sendBodyAndHeader("direct:start", "Hello beer", "where", "beer");
// there should only be one amqp endpoint
- long count = context.getEndpoints().stream().filter(e -> e.getEndpointUri().startsWith("amqp:")).count();
+ long count = contextExtension.getContext().getEndpoints().stream().filter(e -> e.getEndpointUri().startsWith("amqp:"))
+ .count();
assertEquals(1, count, "There should only be 1 amqp endpoint");
// and the messages should be in the queues
@@ -41,15 +50,30 @@ public class AMQPToDSendDynamicTest extends AMQPTestSupport {
assertEquals("Hello beer", out);
}
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext camelContext = super.createCamelContext();
+ @BeforeAll
+ static void startContext() throws Exception {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
+ }
+
+ @BeforeEach
+ void setupTemplate() {
+ template = contextExtension.getProducerTemplate();
+ consumer = contextExtension.getConsumerTemplate();
+ }
+
+ @ContextFixture
+ public void configureContext(CamelContext camelContext) {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
+
camelContext.getRegistry().bind("amqpConnection", discoverAMQP(camelContext));
- return camelContext;
}
- @Override
- protected RouteBuilder createRouteBuilder() {
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
+ private RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
// route message dynamic using toD
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDTest.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDTest.java
index 8243a637e75..29417634520 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDTest.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/AMQPToDTest.java
@@ -17,34 +17,54 @@
package org.apache.camel.component.amqp;
import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.RouteFixture;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.camel.component.amqp.AMQPConnectionDetails.discoverAMQP;
public class AMQPToDTest extends AMQPTestSupport {
+ private ProducerTemplate template;
@Test
public void testToD() throws Exception {
- getMockEndpoint("mock:bar").expectedBodiesReceived("Hello bar");
- getMockEndpoint("mock:beer").expectedBodiesReceived("Hello beer");
+ contextExtension.getMockEndpoint("mock:bar").expectedBodiesReceived("Hello bar");
+ contextExtension.getMockEndpoint("mock:beer").expectedBodiesReceived("Hello beer");
template.sendBodyAndHeader("direct:start", "Hello bar", "where", "bar");
template.sendBodyAndHeader("direct:start", "Hello beer", "where", "beer");
- MockEndpoint.assertIsSatisfied(context);
+ MockEndpoint.assertIsSatisfied(contextExtension.getContext());
}
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext camelContext = super.createCamelContext();
+ @BeforeAll
+ static void startContext() throws Exception {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
+ }
+
+ @BeforeEach
+ void setupTemplate() {
+ template = contextExtension.getProducerTemplate();
+ }
+
+ @ContextFixture
+ public void configureContext(CamelContext camelContext) {
+ System.setProperty(AMQPConnectionDetails.AMQP_PORT, service.brokerPort() + "");
+
camelContext.getRegistry().bind("amqpConnection", discoverAMQP(camelContext));
- return camelContext;
}
- @Override
- protected RouteBuilder createRouteBuilder() {
+ @RouteFixture
+ public void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
+ }
+
+ private RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
// route message dynamic using toD
diff --git a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
index 5654f4b5c1b..8ae8d956223 100644
--- a/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
+++ b/components/camel-amqp/src/test/java/org/apache/camel/component/amqp/artemis/AMQPEmbeddedBrokerTest.java
@@ -19,15 +19,17 @@ package org.apache.camel.component.amqp.artemis;
import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
-import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.amqp.AMQPComponent;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.infra.artemis.services.ArtemisService;
import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeAll;
+import org.apache.camel.test.infra.core.CamelContextExtension;
+import org.apache.camel.test.infra.core.ContextFixture;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.apache.camel.test.infra.core.RouteFixture;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -35,26 +37,26 @@ import static org.apache.camel.component.amqp.AMQPConnectionDetails.AMQP_PORT;
import static org.apache.camel.component.amqp.AMQPConnectionDetails.AMQP_SET_TOPIC_PREFIX;
import static org.apache.camel.component.amqp.AMQPConnectionDetails.discoverAMQP;
-public class AMQPEmbeddedBrokerTest extends CamelTestSupport {
+public class AMQPEmbeddedBrokerTest {
+ @Order(1)
@RegisterExtension
- static ArtemisService service = ArtemisServiceFactory.createSingletonAMQPService();
+ public static ArtemisService service = ArtemisServiceFactory.createSingletonAMQPService();
- @EndpointInject("mock:result")
- MockEndpoint resultEndpoint;
+ @Order(2)
+ @RegisterExtension
+ public static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
- String expectedBody = "Hello there!";
+ private MockEndpoint resultEndpoint;
- @BeforeAll
- public static void beforeClass() throws Exception {
- System.setProperty(AMQP_PORT, service.brokerPort() + "");
- System.setProperty(AMQP_SET_TOPIC_PREFIX, "false");
- }
+ private String expectedBody = "Hello there!";
@BeforeEach
void prepareTest() {
+ resultEndpoint = contextExtension.getMockEndpoint("mock:result");
+
resultEndpoint.expectedMessageCount(1);
- template.sendBody("direct:send-topic", expectedBody);
+ contextExtension.getProducerTemplate().sendBody("direct:send-topic", expectedBody);
}
@Test
@@ -62,16 +64,21 @@ public class AMQPEmbeddedBrokerTest extends CamelTestSupport {
resultEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS);
}
- @Override
- protected CamelContext createCamelContext() throws Exception {
- CamelContext camelContext = super.createCamelContext();
- camelContext.getRegistry().bind("amqpConnection", discoverAMQP(camelContext));
- camelContext.addComponent("amqp-customized", new AMQPComponent());
- return camelContext;
+ @ContextFixture
+ public static void setupRoutes(CamelContext context) throws Exception {
+ System.setProperty(AMQP_PORT, service.brokerPort() + "");
+ System.setProperty(AMQP_SET_TOPIC_PREFIX, "false");
+
+ context.getRegistry().bind("amqpConnection", discoverAMQP(context));
+ context.addComponent("amqp-customized", new AMQPComponent());
+ }
+
+ @RouteFixture
+ public static void createRouteBuilder(CamelContext context) throws Exception {
+ context.addRoutes(createRouteBuilder());
}
- @Override
- protected RouteBuilder createRouteBuilder() {
+ private static RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
from("direct:send-topic")