This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 71f546438529 CAMEL-26618: camel-azure-eventhubs - run the integration
tests on Floci AZ (#29245)
71f546438529 is described below
commit 71f546438529969384462a6343d82f9963372a09
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Nov 9 13:41:30 2026 +0100
CAMEL-26618: camel-azure-eventhubs - run the integration tests on Floci AZ
(#29245)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../catalog/docs/azure-eventhubs-component.adoc | 12 +-
.../apache/camel/catalog/test-infra/metadata.json | 11 ++
.../camel-azure/camel-azure-eventhubs/pom.xml | 9 +
.../src/main/docs/azure-eventhubs-component.adoc | 12 +-
.../camel/component/azure/eventhubs/TestUtils.java | 39 ++--
.../eventhubs/integration/EventHubsConsumerIT.java | 43 +++--
.../EventHubsProducerCustomClientIT.java | 14 +-
.../eventhubs/integration/EventHubsProducerIT.java | 17 +-
.../eventhubs/integration/EventProcessorIT.java | 30 +--
.../operations/EventHubsProducerOperationsIT.java | 19 +-
test-infra/camel-test-infra-all/pom.xml | 11 ++
.../src/generated/resources/META-INF/metadata.json | 11 ++
.../test/infra/azure/common/AzureProperties.java | 6 +
.../azure/common/services/FlociAzContainer.java | 109 +++++++++++
.../azure/common/services/container.properties | 6 +
.../camel-test-infra-azure-eventhubs/pom.xml | 58 ++++++
.../clients/AzureEventHubsClientUtils.java | 44 +++++
.../eventhubs/common/AzureEventHubsProperties.java | 39 ++++
.../services/AzureEventHubsInfraService.java} | 31 ++--
.../AzureEventHubsLocalContainerInfraService.java | 201 +++++++++++++++++++++
.../services/AzureEventHubsRemoteInfraService.java | 62 +++++++
.../eventhubs/services/AzureEventHubsService.java} | 21 +--
.../services/AzureEventHubsServiceFactory.java | 92 ++++++++++
.../common/AzureServiceBusProperties.java | 4 -
.../AzureServiceBusLocalContainerInfraService.java | 99 ++--------
.../azure/servicebus/services/container.properties | 22 ---
test-infra/pom.xml | 1 +
27 files changed, 823 insertions(+), 200 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-eventhubs-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-eventhubs-component.adoc
index 0ac28e852335..716fda655a0d 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-eventhubs-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-eventhubs-component.adoc
@@ -223,9 +223,17 @@ from("direct:start")
== Important Development Notes
-When developing on this component, you will need to obtain your Azure
accessKey to run the integration tests. In addition to the mocked unit tests,
+When developing on this component, in addition to the mocked unit tests,
you *will need to run the integration tests with every change you make or even
client upgrade as the Azure client can break things even on minor versions
upgrade.*
-To run the integration tests, on this component directory, run the following
maven command:
+The integration tests run against a local
https://github.com/floci-io/floci-az[Floci AZ] container, which needs Docker.
+To run them, on this component directory, run the following maven command:
+
+[source,bash]
+----
+mvn verify
+----
+
+To run the integration tests against an event hub in Azure, run the following
maven command instead:
[source,bash]
----
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/test-infra/metadata.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/test-infra/metadata.json
index 963e52191569..cc8303394261 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/test-infra/metadata.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/test-infra/metadata.json
@@ -251,6 +251,17 @@
"version" : "4.24.0-SNAPSHOT",
"serviceVersion" : "3.37.0",
"uiSupported" : false
+}, {
+ "service" :
"org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsInfraService",
+ "description" : "Local Azure Event Hubs service with Floci AZ",
+ "implementation" :
"org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsLocalContainerInfraService",
+ "alias" : [ "azure" ],
+ "aliasImplementation" : [ "event-hubs" ],
+ "groupId" : "org.apache.camel",
+ "artifactId" : "camel-test-infra-azure-eventhubs",
+ "version" : "4.24.0-SNAPSHOT",
+ "serviceVersion" : "0.16.0",
+ "uiSupported" : false
}, {
"service" :
"org.apache.camel.test.infra.infinispan.services.InfinispanInfraService",
"description" : "Infinispan is a distributed in-memory key/value data store",
diff --git a/components/camel-azure/camel-azure-eventhubs/pom.xml
b/components/camel-azure/camel-azure-eventhubs/pom.xml
index ce7e502be543..d8276b50aaf2 100644
--- a/components/camel-azure/camel-azure-eventhubs/pom.xml
+++ b/components/camel-azure/camel-azure-eventhubs/pom.xml
@@ -35,6 +35,9 @@
<properties>
<!-- the DST tests load the byte code agent of camel-test-dst when the
JVM starts (see test-dst.adoc) -->
<camel.surefire.fork.additional-vmargs>"-javaagent:${net.bytebuddy:byte-buddy-agent:jar}"</camel.surefire.fork.additional-vmargs>
+ <!-- Floci AZ is not available on these platforms -->
+ <skipITs.ppc64le>true</skipITs.ppc64le>
+ <skipITs.s390x>true</skipITs.s390x>
</properties>
<dependencies>
@@ -89,6 +92,12 @@
<artifactId>camel-test-dst</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-azure-eventhubs</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
b/components/camel-azure/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
index 0ac28e852335..716fda655a0d 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
@@ -223,9 +223,17 @@ from("direct:start")
== Important Development Notes
-When developing on this component, you will need to obtain your Azure
accessKey to run the integration tests. In addition to the mocked unit tests,
+When developing on this component, in addition to the mocked unit tests,
you *will need to run the integration tests with every change you make or even
client upgrade as the Azure client can break things even on minor versions
upgrade.*
-To run the integration tests, on this component directory, run the following
maven command:
+The integration tests run against a local
https://github.com/floci-io/floci-az[Floci AZ] container, which needs Docker.
+To run them, on this component directory, run the following maven command:
+
+[source,bash]
+----
+mvn verify
+----
+
+To run the integration tests against an event hub in Azure, run the following
maven command instead:
[source,bash]
----
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
index af7d9def72b7..07ef07975741 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
@@ -19,6 +19,9 @@ package org.apache.camel.component.azure.eventhubs;
import java.io.IOException;
import java.util.Properties;
+import
org.apache.camel.test.infra.azure.eventhubs.common.AzureEventHubsProperties;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsServiceFactory;
import org.apache.camel.test.junit6.TestSupport;
public final class TestUtils {
@@ -34,20 +37,32 @@ public final class TestUtils {
return TestSupport.loadExternalProperties(TestUtils.class,
"azure_key.properties");
}
- public static Properties loadAzureAccessFromJvmEnv() throws Exception {
- final Properties properties = new Properties();
- if (System.getProperty(CONNECTION_STRING) == null
- || System.getProperty(BLOB_ACCOUNT_NAME) == null
- || System.getProperty(BLOB_ACCESS_KEY) == null) {
- throw new Exception(
- "Make sure to supply azure eventHubs connectionString,
e.g: mvn verify -DconnectionString=string"
- + " -DblobAccountName=blob
-DblobAccessKey=key");
+ /**
+ * The service of the integration tests: an event hub on a local Floci AZ
container, unless the connectionString,
+ * blobAccountName and blobAccessKey system properties are set: then the
event hub of that connection string, with
+ * the checkpoints in that storage account.
+ */
+ public static AzureEventHubsService createService() {
+ String connectionString = System.getProperty(CONNECTION_STRING);
+ if (connectionString != null) {
+ if (System.getProperty(BLOB_ACCOUNT_NAME) == null ||
System.getProperty(BLOB_ACCESS_KEY) == null) {
+ throw new IllegalArgumentException(
+ "Make sure to supply azure eventHubs connectionString,
e.g: mvn verify -DconnectionString=string"
+ + " -DblobAccountName=blob
-DblobAccessKey=key");
+ }
+ System.setProperty(AzureEventHubsProperties.CONNECTION_STRING,
connectionString);
+ System.setProperty(AzureEventHubsProperties.BLOB_ACCOUNT_NAME,
System.getProperty(BLOB_ACCOUNT_NAME));
+ System.setProperty(AzureEventHubsProperties.BLOB_ACCESS_KEY,
System.getProperty(BLOB_ACCESS_KEY));
+ System.setProperty("azure.instance.type", "remote");
}
- properties.setProperty(CONNECTION_STRING,
System.getProperty(CONNECTION_STRING));
- properties.setProperty(BLOB_ACCOUNT_NAME,
System.getProperty(BLOB_ACCOUNT_NAME));
- properties.setProperty(BLOB_ACCESS_KEY,
System.getProperty(BLOB_ACCESS_KEY));
- return properties;
+ return AzureEventHubsServiceFactory.createSingletonService();
}
+ /**
+ * Whether the integration tests run against Azure rather than a local
Floci AZ container
+ */
+ public static boolean isRemote() {
+ return "remote".equals(System.getProperty("azure.instance.type"));
+ }
}
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsConsumerIT.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsConsumerIT.java
index 209ef537d637..bc68a8c2a4a3 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsConsumerIT.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsConsumerIT.java
@@ -21,10 +21,10 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Properties;
import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.messaging.eventhubs.models.EventPosition;
import com.azure.messaging.eventhubs.models.SendOptions;
import com.azure.storage.blob.BlobContainerAsyncClient;
@@ -37,22 +37,25 @@ import
org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.azure.eventhubs.TestUtils;
import
org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
import org.apache.camel.component.mock.MockEndpoint;
+import
org.apache.camel.test.infra.azure.eventhubs.clients.AzureEventHubsClientUtils;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
-@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
- disabledReason = "Make sure to supply azure eventHubs
connectionString, e.g: mvn verify -DconnectionString=string
-DblobAccountName=blob -DblobAccessKey=key")
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EventHubsConsumerIT extends CamelTestSupport {
+ @RegisterExtension
+ static AzureEventHubsService service = TestUtils.createService();
+
@EndpointInject("mock:result")
private MockEndpoint result;
@@ -61,18 +64,13 @@ class EventHubsConsumerIT extends CamelTestSupport {
private EventHubsConfiguration configuration;
@BeforeAll
- public void prepare() throws Exception {
+ public void prepare() {
containerName =
RandomStringUtils.secure().nextAlphabetic(5).toLowerCase();
- final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
-
configuration = new EventHubsConfiguration();
-
configuration.setBlobAccessKey(properties.getProperty(TestUtils.BLOB_ACCESS_KEY));
-
configuration.setBlobAccountName(properties.getProperty(TestUtils.BLOB_ACCOUNT_NAME));
- configuration.setBlobContainerName(containerName);
-
configuration.setConnectionString(properties.getProperty(TestUtils.CONNECTION_STRING));
+ configuration.setConnectionString(service.connectionString());
- containerAsyncClient =
EventHubsClientFactory.createBlobContainerClient(configuration);
+ containerAsyncClient =
AzureEventHubsClientUtils.createBlobContainerClient(service, containerName);
// create test container
containerAsyncClient.create().block();
@@ -129,9 +127,8 @@ class EventHubsConsumerIT extends CamelTestSupport {
@Override
public void configure() {
from("azure-eventhubs:?"
- + "connectionString=RAW({{connectionString}})"
- + "&blobContainerName=" + containerName +
"&eventPosition=#eventPosition"
- +
"&blobAccountName={{blobAccountName}}&blobAccessKey=RAW({{blobAccessKey}})")
+ + "connectionString=RAW(" + service.connectionString() +
")"
+ + "&eventPosition=#eventPosition" + checkpointOptions())
.to(result);
}
@@ -146,7 +143,23 @@ class EventHubsConsumerIT extends CamelTestSupport {
CamelContext context = super.createCamelContext();
context.getRegistry().bind("eventPosition", positionMap);
+ if (!TestUtils.isRemote()) {
+ context.getRegistry().bind("checkpointStore", new
BlobCheckpointStore(containerAsyncClient));
+ }
return context;
}
+
+ /*
+ * Against Azure the consumer builds its checkpoint store from the Blob
options, as an application does. A local
+ * service has its Blob storage at an endpoint of its own, which the
component builds no client for, so the consumer
+ * gets a checkpoint store on that endpoint.
+ */
+ private String checkpointOptions() {
+ if (TestUtils.isRemote()) {
+ return "&blobContainerName=" + containerName + "&blobAccountName="
+ service.blobAccountName()
+ + "&blobAccessKey=RAW(" + service.blobAccessKey() + ")";
+ }
+ return "&checkpointStore=#checkpointStore";
+ }
}
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerCustomClientIT.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerCustomClientIT.java
index 62471697c986..9d4f45e7b6ac 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerCustomClientIT.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerCustomClientIT.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.azure.eventhubs.integration;
-import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -34,19 +33,21 @@ import
org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.azure.eventhubs.TestUtils;
import
org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
import org.apache.camel.component.mock.MockEndpoint;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
- disabledReason = "Make sure to supply azure eventHubs
connectionString, e.g: mvn verify -DconnectionString=string
-DblobAccountName=blob -DblobAccessKey=key")
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EventHubsProducerCustomClientIT extends CamelTestSupport {
+ @RegisterExtension
+ static AzureEventHubsService service = TestUtils.createService();
+
@EndpointInject
private ProducerTemplate template;
@@ -59,10 +60,9 @@ class EventHubsProducerCustomClientIT extends
CamelTestSupport {
private EventHubConsumerAsyncClient consumerAsyncClient;
@BeforeAll
- public void prepare() throws Exception {
- final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+ public void prepare() {
final EventHubsConfiguration configuration = new
EventHubsConfiguration();
-
configuration.setConnectionString(properties.getProperty(TestUtils.CONNECTION_STRING));
+ configuration.setConnectionString(service.connectionString());
configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
consumerAsyncClient =
EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerIT.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerIT.java
index cdb2bf232e0d..4c8ce885c6ed 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerIT.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventHubsProducerIT.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.azure.eventhubs.integration;
-import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -33,22 +32,24 @@ import
org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.azure.eventhubs.TestUtils;
import
org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
import org.apache.camel.component.mock.MockEndpoint;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
- disabledReason = "Make sure to supply azure eventHubs
connectionString, e.g: mvn verify -DconnectionString=string
-DblobAccountName=blob -DblobAccessKey=key")
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EventHubsProducerIT extends CamelTestSupport {
+ @RegisterExtension
+ static AzureEventHubsService service = TestUtils.createService();
+
@EndpointInject
private ProducerTemplate template;
@@ -58,11 +59,9 @@ class EventHubsProducerIT extends CamelTestSupport {
private EventHubConsumerAsyncClient consumerAsyncClient;
@BeforeAll
- public void prepare() throws Exception {
- final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
-
+ public void prepare() {
final EventHubsConfiguration configuration = new
EventHubsConfiguration();
-
configuration.setConnectionString(properties.getProperty(TestUtils.CONNECTION_STRING));
+ configuration.setConnectionString(service.connectionString());
configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
consumerAsyncClient =
EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
@@ -137,7 +136,7 @@ class EventHubsProducerIT extends CamelTestSupport {
@Override
public void configure() {
from("direct:sendAsync")
-
.to("azure-eventhubs:?connectionString=RAW({{connectionString}})")
+ .to("azure-eventhubs:?connectionString=RAW(" +
service.connectionString() + ")")
.to(result);
}
};
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventProcessorIT.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventProcessorIT.java
index 3f253cc3e6e9..8cb3302c4006 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventProcessorIT.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/EventProcessorIT.java
@@ -19,7 +19,6 @@ package
org.apache.camel.component.azure.eventhubs.integration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
@@ -28,6 +27,7 @@ import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventHubClientBuilder;
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
import com.azure.messaging.eventhubs.EventProcessorClient;
+import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.messaging.eventhubs.models.ErrorContext;
import com.azure.messaging.eventhubs.models.EventContext;
import com.azure.messaging.eventhubs.models.EventPosition;
@@ -36,42 +36,50 @@ import com.azure.storage.blob.BlobContainerAsyncClient;
import org.apache.camel.component.azure.eventhubs.EventHubsConfiguration;
import org.apache.camel.component.azure.eventhubs.TestUtils;
import
org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import
org.apache.camel.test.infra.azure.eventhubs.clients.AzureEventHubsClientUtils;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
import org.apache.commons.lang3.RandomStringUtils;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
- disabledReason = "Make sure to supply azure eventHubs
connectionString, e.g: mvn verify -DconnectionString=string
-DblobAccountName=blob -DblobAccessKey=key")
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class EventProcessorIT {
+ @RegisterExtension
+ static AzureEventHubsService service = TestUtils.createService();
+
private EventHubsConfiguration configuration;
private BlobContainerAsyncClient containerAsyncClient;
@BeforeAll
- public void prepare() throws Exception {
- final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+ public void prepare() {
final String containerName =
RandomStringUtils.secure().nextAlphabetic(5).toLowerCase();
configuration = new EventHubsConfiguration();
-
configuration.setConnectionString(properties.getProperty("connectionString"));
+ configuration.setConnectionString(service.connectionString());
configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
-
configuration.setBlobAccessKey(properties.getProperty("blobAccessKey"));
-
configuration.setBlobAccountName(properties.getProperty("blobAccountName"));
Map<String, EventPosition> positionMap = new HashMap<>();
positionMap.put("0", EventPosition.earliest());
configuration.setEventPosition(positionMap);
- configuration.setBlobContainerName(containerName);
- containerAsyncClient =
EventHubsClientFactory.createBlobContainerClient(configuration);
+ containerAsyncClient =
AzureEventHubsClientUtils.createBlobContainerClient(service, containerName);
+ if (TestUtils.isRemote()) {
+ // against Azure the event processor builds its checkpoint store
from the Blob options, as an application does
+ configuration.setBlobAccountName(service.blobAccountName());
+ configuration.setBlobAccessKey(service.blobAccessKey());
+ configuration.setBlobContainerName(containerName);
+ } else {
+ // a local service has its Blob storage at an endpoint of its own,
which the component builds no client for
+ configuration.setCheckpointStore(new
BlobCheckpointStore(containerAsyncClient));
+ }
// create test container
containerAsyncClient.create().block();
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java
index 80786f835149..1e6061227227 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java
@@ -19,7 +19,6 @@ package
org.apache.camel.component.azure.eventhubs.integration.operations;
import java.time.Duration;
import java.util.LinkedList;
import java.util.List;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
import com.azure.messaging.eventhubs.EventHubClientBuilder;
@@ -33,29 +32,29 @@ import org.apache.camel.component.azure.eventhubs.TestUtils;
import
org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
import
org.apache.camel.component.azure.eventhubs.operations.EventHubsProducerOperations;
import org.apache.camel.support.DefaultExchange;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsService;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
-@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
- disabledReason = "Make sure to supply azure eventHubs
connectionString, e.g: mvn verify -DconnectionString=string
-DblobAccountName=blob -DblobAccessKey=key")
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EventHubsProducerOperationsIT extends CamelTestSupport {
+ @RegisterExtension
+ static AzureEventHubsService service = TestUtils.createService();
+
private EventHubsConfiguration configuration;
private EventHubProducerAsyncClient producerAsyncClient;
private EventHubConsumerAsyncClient consumerAsyncClient;
@BeforeAll
- public void prepare() throws Exception {
- final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
-
+ public void prepare() {
configuration = new EventHubsConfiguration();
-
configuration.setConnectionString(properties.getProperty("connectionString"));
+ configuration.setConnectionString(service.connectionString());
configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
producerAsyncClient =
EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
@@ -112,6 +111,8 @@ class EventHubsProducerOperationsIT extends
CamelTestSupport {
exchanges.add(exchange2);
final Exchange exchange = new DefaultExchange(context);
+ // the events are looked for in this partition only
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID,
firstPartition);
exchange.getIn().setBody(exchanges);
operations.sendEvents(exchange, doneSync -> {
@@ -155,6 +156,8 @@ class EventHubsProducerOperationsIT extends
CamelTestSupport {
messages.add("Test String Message 2");
final Exchange exchange = new DefaultExchange(context);
+ // the events are looked for in this partition only
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID,
firstPartition);
exchange.getIn().setBody(messages);
operations.sendEvents(exchange, doneSync -> {
diff --git a/test-infra/camel-test-infra-all/pom.xml
b/test-infra/camel-test-infra-all/pom.xml
index d1725e0352e9..b01eac332f60 100644
--- a/test-infra/camel-test-infra-all/pom.xml
+++ b/test-infra/camel-test-infra-all/pom.xml
@@ -76,6 +76,11 @@
<artifactId>camel-test-infra-azure-eventgrid</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-azure-eventhubs</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test-infra-azure-functions</artifactId>
@@ -444,6 +449,12 @@
<artifactId>camel-test-infra-azure-eventgrid</artifactId>
</dependency>
</fileSet>
+ <fileSet>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+
<artifactId>camel-test-infra-azure-eventhubs</artifactId>
+ </dependency>
+ </fileSet>
<fileSet>
<dependency>
<groupId>org.apache.camel</groupId>
diff --git
a/test-infra/camel-test-infra-all/src/generated/resources/META-INF/metadata.json
b/test-infra/camel-test-infra-all/src/generated/resources/META-INF/metadata.json
index 963e52191569..cc8303394261 100644
---
a/test-infra/camel-test-infra-all/src/generated/resources/META-INF/metadata.json
+++
b/test-infra/camel-test-infra-all/src/generated/resources/META-INF/metadata.json
@@ -251,6 +251,17 @@
"version" : "4.24.0-SNAPSHOT",
"serviceVersion" : "3.37.0",
"uiSupported" : false
+}, {
+ "service" :
"org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsInfraService",
+ "description" : "Local Azure Event Hubs service with Floci AZ",
+ "implementation" :
"org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsLocalContainerInfraService",
+ "alias" : [ "azure" ],
+ "aliasImplementation" : [ "event-hubs" ],
+ "groupId" : "org.apache.camel",
+ "artifactId" : "camel-test-infra-azure-eventhubs",
+ "version" : "4.24.0-SNAPSHOT",
+ "serviceVersion" : "0.16.0",
+ "uiSupported" : false
}, {
"service" :
"org.apache.camel.test.infra.infinispan.services.InfinispanInfraService",
"description" : "Infinispan is a distributed in-memory key/value data store",
diff --git
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
b/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
index 74f40e210c7b..7f409b69edb7 100644
---
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
+++
b/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
@@ -31,6 +31,12 @@ public final class AzureProperties {
* The Azurite image of the local Queue service
*/
public static final String AZURE_STORAGE_QUEUE_CONTAINER =
"azure.storage.queue.container";
+ /**
+ * The Artemis image of the AMQP brokers that Floci AZ starts for Event
Hubs and Service Bus. Floci AZ copies AMQP
+ * libraries patched for one Artemis version into it, so the version must
be the one that the Floci AZ image of
+ * {@link #AZURE_CONTAINER} expects.
+ */
+ public static final String AZURE_BROKER_CONTAINER =
"azure.broker.container";
private AzureProperties() {
diff --git
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/services/FlociAzContainer.java
b/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/services/FlociAzContainer.java
index 5ce38aca62d2..76839acdf508 100644
---
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/services/FlociAzContainer.java
+++
b/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/services/FlociAzContainer.java
@@ -18,13 +18,31 @@
package org.apache.camel.test.infra.azure.common.services;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.CreateContainerCmd;
+import com.github.dockerjava.api.exception.NotFoundException;
+import com.github.dockerjava.api.model.Bind;
+import com.github.dockerjava.api.model.Container;
+import com.github.dockerjava.api.model.HostConfig;
+import com.github.dockerjava.api.model.Volume;
import org.apache.camel.test.infra.azure.common.AzureCredentialsHolder;
import org.apache.camel.test.infra.azure.common.AzureProperties;
import org.apache.camel.test.infra.common.LocalPropertyResolver;
import org.apache.camel.test.infra.common.services.ContainerEnvironmentUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.RemoteDockerImage;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.ResourceReaper;
/**
* A container running Floci AZ (https://github.com/floci-io/floci-az), a
local Azure emulator. It serves every Azure
@@ -38,6 +56,14 @@ public class FlociAzContainer extends
GenericContainer<FlociAzContainer> {
public static final String DEFAULT_ACCOUNT_KEY
=
"Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+ private static final Logger LOG =
LoggerFactory.getLogger(FlociAzContainer.class);
+ // Floci AZ puts this namespace in the name of the containers it starts,
and in their floci_namespace label
+ private static final String RESOURCE_NAMESPACE_ENV =
"FLOCI_AZ_DOCKER_RESOURCE_NAMESPACE";
+ private static final String RESOURCE_NAMESPACE_LABEL = "floci_namespace";
+ private static final String DOCKER_SOCKET = "/var/run/docker.sock";
+
+ private String resourceNamespace;
+
public FlociAzContainer() {
this(LocalPropertyResolver.getProperty(FlociAzContainer.class,
AzureProperties.AZURE_CONTAINER), false);
}
@@ -81,4 +107,87 @@ public class FlociAzContainer extends
GenericContainer<FlociAzContainer> {
public String accountEndpoint() {
return serviceAddress() + "/" + DEFAULT_ACCOUNT_NAME;
}
+
+ /**
+ * Lets Floci AZ start containers of its own, such as the Artemis brokers
of the AMQP data plane of Event Hubs and
+ * Service Bus: mounts the Docker socket of the host, and gives the
containers that Floci AZ starts a resource
+ * namespace of this container, by which {@link #stop()} removes them.
Each container has a namespace of its own:
+ * when it starts, Floci AZ removes the containers of its namespace that
it finds, as left over.
+ */
+ public FlociAzContainer withDockerSocket() {
+ resourceNamespace = "camel" +
UUID.randomUUID().toString().replace("-", "").substring(0, 12);
+ withEnv(RESOURCE_NAMESPACE_ENV, resourceNamespace);
+ return
withCreateContainerCmdModifier(FlociAzContainer::bindDockerSocket);
+ }
+
+ /**
+ * Has Floci AZ start an AMQP broker with the Artemis image of {@link
AzureProperties#AZURE_BROKER_CONTAINER}, which
+ * this method pulls as Testcontainers pulls its own images, so that image
substitution and registry authentication
+ * apply: Floci AZ would pull it itself otherwise. Needs {@link
#withDockerSocket()}.
+ *
+ * @param imageVariable the variable of Floci AZ that names the broker
image of a service, for example
+ * {@code FLOCI_AZ_SERVICES_EVENT_HUB_ARTEMIS_IMAGE}
+ */
+ public FlociAzContainer withBrokerImage(String imageVariable) {
+ String image =
LocalPropertyResolver.getProperty(FlociAzContainer.class,
AzureProperties.AZURE_BROKER_CONTAINER);
+ return withEnv(imageVariable, new
RemoteDockerImage(DockerImageName.parse(image)).get());
+ }
+
+ @Override
+ public void start() {
+ if (resourceNamespace != null) {
+ // Testcontainers does not know the containers that Floci AZ
starts: have them removed when the JVM exits,
+ // should this container not be stopped
+
ResourceReaper.instance().registerLabelsFilterForCleanup(Map.of(RESOURCE_NAMESPACE_LABEL,
resourceNamespace));
+ }
+ super.start();
+ }
+
+ @Override
+ public void stop() {
+ try {
+ super.stop();
+ } finally {
+ if (resourceNamespace != null) {
+ removeStartedContainers();
+ }
+ }
+ }
+
+ // the containers that Floci AZ started are not removed with it
+ private void removeStartedContainers() {
+ DockerClient client = DockerClientFactory.instance().client();
+ List<Container> containers = client.listContainersCmd()
+ .withShowAll(true)
+ .withLabelFilter(Map.of(RESOURCE_NAMESPACE_LABEL,
resourceNamespace))
+ .exec();
+ for (Container container : containers) {
+ LOG.debug("Removing the container {} that Floci AZ started",
Arrays.toString(container.getNames()));
+ try {
+
client.removeContainerCmd(container.getId()).withForce(true).withRemoveVolumes(true).exec();
+ } catch (NotFoundException e) {
+ LOG.debug("The container {} was already removed",
container.getId());
+ }
+ }
+ }
+
+ // where SELinux is enforcing, it denies a container the Docker socket of
the host unless the labeling of the
+ // container is disabled
+ private static void bindDockerSocket(CreateContainerCmd cmd) {
+ HostConfig hostConfig = cmd.getHostConfig();
+
+ List<Bind> binds = new ArrayList<>();
+ if (hostConfig.getBinds() != null) {
+ binds.addAll(Arrays.asList(hostConfig.getBinds()));
+ }
+ binds.add(new
Bind(DockerClientFactory.instance().getRemoteDockerUnixSocketPath(), new
Volume(DOCKER_SOCKET)));
+
+ List<String> securityOpts = new ArrayList<>();
+ if (hostConfig.getSecurityOpts() != null) {
+ securityOpts.addAll(hostConfig.getSecurityOpts());
+ }
+ securityOpts.add("label=disable");
+
+ hostConfig.withBinds(binds).withSecurityOpts(securityOpts);
+ }
}
diff --git
a/test-infra/camel-test-infra-azure-common/src/main/resources/org/apache/camel/test/infra/azure/common/services/container.properties
b/test-infra/camel-test-infra-azure-common/src/main/resources/org/apache/camel/test/infra/azure/common/services/container.properties
index 3089b7a8a244..98b65b7da966 100644
---
a/test-infra/camel-test-infra-azure-common/src/main/resources/org/apache/camel/test/infra/azure/common/services/container.properties
+++
b/test-infra/camel-test-infra-azure-common/src/main/resources/org/apache/camel/test/infra/azure/common/services/container.properties
@@ -17,6 +17,12 @@
# Floci AZ: the local Azure services
azure.container=mirror.gcr.io/floci/floci-az:0.16.0
azure.container.version.exclude=latest,nightly,compat
+# Artemis: the AMQP broker that Floci AZ starts for Event Hubs and Service
Bus. Floci AZ copies AMQP jars patched for one
+# Artemis version into it, so the version must be the one that the Floci AZ
image of azure.container expects: it is
+# pinned, and changes with azure.container. The key does not end with
"artemis": the test-infra metadata would take
+# this version for the Artemis services.
+azure.broker.container=mirror.gcr.io/apache/activemq-artemis:2.44.0
+azure.broker.container.version.include=2.44.0
# Azurite: the local Blob and Queue services, until Floci AZ covers what their
tests use
azure.storage.blob.container=mcr.microsoft.com/azure-storage/azurite:3.37.0
azure.storage.blob.container.version.exclude=table-alpha,amd64,arm64,preview,latest
diff --git a/test-infra/camel-test-infra-azure-eventhubs/pom.xml
b/test-infra/camel-test-infra-azure-eventhubs/pom.xml
new file mode 100644
index 000000000000..d059f886c9a1
--- /dev/null
+++ b/test-infra/camel-test-infra-azure-eventhubs/pom.xml
@@ -0,0 +1,58 @@
+<?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";>
+ <parent>
+ <artifactId>camel-test-infra-parent</artifactId>
+ <groupId>org.apache.camel</groupId>
+ <relativePath>../camel-test-infra-parent/pom.xml</relativePath>
+ <version>4.24.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <name>Camel :: Test Infra :: Azure Event Hubs</name>
+ <artifactId>camel-test-infra-azure-eventhubs</artifactId>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-sdk-bom</artifactId>
+ <version>${azure-sdk-bom-version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-azure-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- azure sdk: the Blob container of the checkpoint store -->
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git
a/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/clients/AzureEventHubsClientUtils.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/clients/AzureEventHubsClientUtils.java
new file mode 100644
index 000000000000..b9c68efdeb79
--- /dev/null
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/clients/AzureEventHubsClientUtils.java
@@ -0,0 +1,44 @@
+/*
+ * 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.azure.eventhubs.clients;
+
+import com.azure.storage.blob.BlobContainerAsyncClient;
+import com.azure.storage.blob.BlobContainerClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import
org.apache.camel.test.infra.azure.eventhubs.services.AzureEventHubsInfraService;
+
+public final class AzureEventHubsClientUtils {
+
+ private AzureEventHubsClientUtils() {
+
+ }
+
+ /**
+ * A client of a Blob container in the storage account of the service, for
example for the
+ * {@code BlobCheckpointStore} of an event processor. A
camel-azure-eventhubs consumer builds the Blob client of its
+ * checkpoint store for the endpoint of an account in Azure only, so a
local service needs a checkpoint store built
+ * on this client.
+ */
+ public static BlobContainerAsyncClient
createBlobContainerClient(AzureEventHubsInfraService service, String
containerName) {
+ return new BlobContainerClientBuilder()
+ .endpoint(service.blobEndpoint())
+ .containerName(containerName)
+ .credential(new
StorageSharedKeyCredential(service.blobAccountName(), service.blobAccessKey()))
+ .buildAsyncClient();
+ }
+}
diff --git
a/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/common/AzureEventHubsProperties.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/common/AzureEventHubsProperties.java
new file mode 100644
index 000000000000..9f3e6d6d3069
--- /dev/null
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/common/AzureEventHubsProperties.java
@@ -0,0 +1,39 @@
+/*
+ * 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.azure.eventhubs.common;
+
+/**
+ * The system properties of the Event Hubs service: the local service sets
them, the remote service reads them.
+ */
+public final class AzureEventHubsProperties {
+ /**
+ * The connection string of the event hub, with the name of the event hub
in its {@code EntityPath}
+ */
+ public static final String CONNECTION_STRING =
"azure.eventhubs.connection.string";
+ /**
+ * The endpoint of the Blob service that holds the checkpoints. A remote
service defaults to
+ * {@code https://<blob account name>.blob.core.windows.net}.
+ */
+ public static final String BLOB_ENDPOINT = "azure.eventhubs.blob.endpoint";
+ public static final String BLOB_ACCOUNT_NAME =
"azure.eventhubs.blob.account.name";
+ public static final String BLOB_ACCESS_KEY =
"azure.eventhubs.blob.access.key";
+
+ private AzureEventHubsProperties() {
+
+ }
+}
diff --git
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsInfraService.java
similarity index 50%
copy from
test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
copy to
test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsInfraService.java
index 74f40e210c7b..aee5f45c64af 100644
---
a/test-infra/camel-test-infra-azure-common/src/main/java/org/apache/camel/test/infra/azure/common/AzureProperties.java
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsInfraService.java
@@ -15,24 +15,29 @@
* limitations under the License.
*/
-package org.apache.camel.test.infra.azure.common;
+package org.apache.camel.test.infra.azure.eventhubs.services;
+
+import org.apache.camel.test.infra.common.services.InfrastructureService;
+
+/**
+ * Test infra service for Azure Event Hubs: an event hub, and the Blob storage
account where an event processor keeps
+ * its checkpoints.
+ */
+public interface AzureEventHubsInfraService extends InfrastructureService {
-public final class AzureProperties {
- /**
- * The Floci AZ image of the local Azure services
- */
- public static final String AZURE_CONTAINER = "azure.container";
/**
- * The Azurite image of the local Blob service. Blob and Queue keep a key
each, so the version of the test-infra
- * metadata is matched by their implementation alias and not by the shared
{@code azure} alias of Floci AZ.
+ * The connection string of the event hub, with the name of the event hub
in its {@code EntityPath}, as the
+ * camel-azure-eventhubs connectionString option expects it.
*/
- public static final String AZURE_STORAGE_BLOB_CONTAINER =
"azure.storage.blob.container";
+ String connectionString();
+
/**
- * The Azurite image of the local Queue service
+ * The endpoint of the Blob service of the storage account that holds the
checkpoints, for example
+ * {@code https://myaccount.blob.core.windows.net}.
*/
- public static final String AZURE_STORAGE_QUEUE_CONTAINER =
"azure.storage.queue.container";
+ String blobEndpoint();
- private AzureProperties() {
+ String blobAccountName();
- }
+ String blobAccessKey();
}
diff --git
a/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsLocalContainerInfraService.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsLocalContainerInfraService.java
new file mode 100644
index 000000000000..b5a71a6e1d68
--- /dev/null
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsLocalContainerInfraService.java
@@ -0,0 +1,201 @@
+/*
+ * 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.azure.eventhubs.services;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.camel.spi.annotations.InfraService;
+import org.apache.camel.test.infra.azure.common.services.FlociAzContainer;
+import org.apache.camel.test.infra.azure.common.services.FlociAzInfraService;
+import
org.apache.camel.test.infra.azure.eventhubs.common.AzureEventHubsProperties;
+import org.apache.camel.test.infra.common.services.ContainerEnvironmentUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A local Event Hubs service with Floci AZ, which runs the AMQP broker of an
Event Hubs namespace in an Apache ActiveMQ
+ * Artemis container that it starts itself, through the Docker socket mounted
into the Floci AZ container. The namespace
+ * has one event hub, {@value #EVENT_HUB_NAME}, with {@value #PARTITION_COUNT}
partitions, and the checkpoints go to the
+ * Blob service of Floci AZ.
+ * <p>
+ * Floci AZ reaches the broker on the Docker network, the clients on the port
Docker publishes it on. Testcontainers
+ * does not know the broker container: stopping the Floci AZ container removes
it (see
+ * {@link FlociAzContainer#withDockerSocket()}).
+ * <p>
+ * Floci AZ answers the {@code $management} requests for the properties of the
event hub, but not for the properties of
+ * a partition. It spreads the events of a batch sent with no partition id or
key over the partitions, one event after
+ * the other, where Azure stores the whole batch in one partition.
+ */
+@InfraService(service = AzureEventHubsInfraService.class,
+ description = "Local Azure Event Hubs service with Floci AZ",
+ serviceAlias = "azure", serviceImplementationAlias =
"event-hubs")
+public class AzureEventHubsLocalContainerInfraService extends
FlociAzInfraService implements AzureEventHubsInfraService {
+ /**
+ * The event hub, in lower case: an event processor of the Azure SDK
lower-cases the name of its event hub
+ */
+ public static final String EVENT_HUB_NAME = "camel-test";
+ public static final int PARTITION_COUNT = 2;
+ /**
+ * The consumer groups of the event hub. Floci AZ takes their names as
case-sensitive, while Azure does not, and an
+ * event processor of the Azure SDK lower-cases the name of its consumer
group: {@code $default} is the default
+ * consumer group of the event processors, {@code $Default} the one of the
other consumers.
+ */
+ public static final List<String> CONSUMER_GROUPS = List.of("$Default",
"$default");
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AzureEventHubsLocalContainerInfraService.class);
+
+ // the namespace that Floci AZ creates by default
+ private static final String NAMESPACE = "emulatorNs1";
+ private static final String ARTEMIS_IMAGE_VARIABLE =
"FLOCI_AZ_SERVICES_EVENT_HUB_ARTEMIS_IMAGE";
+ private static final Pattern AMQP_HOST_PORT =
Pattern.compile("\"amqpHostPort\"\\s*:\\s*(\\d+)");
+ // Floci AZ ignores the key: the broker accepts any client
+ private static final String SHARED_ACCESS_KEY_NAME =
"RootManageSharedAccessKey";
+ private static final String SHARED_ACCESS_KEY = "devkey";
+
+ private int amqpPort;
+ private volatile boolean started;
+
+ @Override
+ protected FlociAzContainer initContainer(String imageName) {
+ FlociAzContainer container = super.initContainer(imageName)
+ .withDockerSocket()
+ .withEnv("FLOCI_AZ_SERVICES_EVENT_HUB_ENTITIES",
EVENT_HUB_NAME + ":" + PARTITION_COUNT)
+ .withEnv("FLOCI_AZ_SERVICES_EVENT_HUB_CONSUMER_GROUPS",
String.join(",", CONSUMER_GROUPS));
+ if (!ContainerEnvironmentUtil.isFixedPort(getClass())) {
+ // Docker publishes the broker on free host ports, so that builds
running at the same time do not clash;
+ // camel infra run keeps the ports 5672 (AMQP) and 5671 (AMQPS)
that Floci AZ documents
+ container.withEnv("FLOCI_AZ_SERVICES_EVENT_HUB_AMQP_PORT", "0")
+ .withEnv("FLOCI_AZ_SERVICES_EVENT_HUB_AMQP_TLS_PORT", "0");
+ }
+ return container;
+ }
+
+ @Override
+ public void initialize() {
+ started = true;
+ try {
+ getContainer().withBrokerImage(ARTEMIS_IMAGE_VARIABLE);
+ getContainer().start();
+ amqpPort = startNamespace();
+ } catch (RuntimeException e) {
+ // also removes the broker, should Floci AZ have started it
+ try {
+ shutdown();
+ } catch (RuntimeException suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+
+ LOG.info("Floci AZ running at address {}, with the Event Hubs broker
at {}:{}", getContainer().serviceAddress(),
+ host(), amqpPort);
+
+ registerProperties();
+ }
+
+ /**
+ * Floci AZ starts the broker of a namespace when the namespace is
created, and the request returns once the broker
+ * listens.
+ *
+ * @return the host port of the AMQP port of the broker
+ */
+ private int startNamespace() {
+ URI uri = URI.create(getContainer().serviceAddress() + "/" +
FlociAzContainer.DEFAULT_ACCOUNT_NAME
+ + "-eventhub/namespaces/" + NAMESPACE);
+ HttpRequest request = HttpRequest.newBuilder(uri)
+ .timeout(Duration.ofMinutes(2))
+ .PUT(HttpRequest.BodyPublishers.noBody())
+ .build();
+
+ HttpResponse<String> response;
+ try (HttpClient client = HttpClient.newHttpClient()) {
+ response = client.send(request,
HttpResponse.BodyHandlers.ofString());
+ } catch (IOException e) {
+ throw new IllegalStateException("Floci AZ did not start the Event
Hubs namespace " + NAMESPACE, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while Floci AZ was
starting the Event Hubs namespace", e);
+ }
+
+ Matcher matcher = AMQP_HOST_PORT.matcher(response.body());
+ if (response.statusCode() / 100 != 2 || !matcher.find()) {
+ throw new IllegalStateException(
+ "Floci AZ did not start the Event Hubs namespace " +
NAMESPACE + ": HTTP " + response.statusCode() + " "
+ + response.body());
+ }
+ return Integer.parseInt(matcher.group(1));
+ }
+
+ /**
+ * Stops Floci AZ, which removes the broker too. A singleton service is
shut down by a JVM shutdown hook for each
+ * test class that used it, and these hooks run at the same time.
+ */
+ @Override
+ public synchronized void shutdown() {
+ if (!started) {
+ return;
+ }
+ started = false;
+
+ super.shutdown();
+ }
+
+ @Override
+ public void registerProperties() {
+ super.registerProperties();
+ System.setProperty(AzureEventHubsProperties.CONNECTION_STRING,
connectionString());
+ System.setProperty(AzureEventHubsProperties.BLOB_ENDPOINT,
blobEndpoint());
+ System.setProperty(AzureEventHubsProperties.BLOB_ACCOUNT_NAME,
blobAccountName());
+ System.setProperty(AzureEventHubsProperties.BLOB_ACCESS_KEY,
blobAccessKey());
+ }
+
+ /**
+ * The connection string of the event hub. {@code
UseDevelopmentEmulator=true} makes the client connect with plain
+ * AMQP.
+ */
+ @Override
+ public String connectionString() {
+ return "Endpoint=sb://" + host() + ":" + amqpPort
+ + ";SharedAccessKeyName=" + SHARED_ACCESS_KEY_NAME
+ + ";SharedAccessKey=" + SHARED_ACCESS_KEY
+ + ";UseDevelopmentEmulator=true;EntityPath=" + EVENT_HUB_NAME;
+ }
+
+ @Override
+ public String blobEndpoint() {
+ return getContainer().accountEndpoint();
+ }
+
+ @Override
+ public String blobAccountName() {
+ return accountName();
+ }
+
+ @Override
+ public String blobAccessKey() {
+ return accessKey();
+ }
+}
diff --git
a/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsRemoteInfraService.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsRemoteInfraService.java
new file mode 100644
index 000000000000..c666990a38f6
--- /dev/null
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsRemoteInfraService.java
@@ -0,0 +1,62 @@
+/*
+ * 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.azure.eventhubs.services;
+
+import
org.apache.camel.test.infra.azure.eventhubs.common.AzureEventHubsProperties;
+
+/**
+ * An event hub and a storage account in Azure, given by the {@link
AzureEventHubsProperties} system properties.
+ */
+public class AzureEventHubsRemoteInfraService implements
AzureEventHubsInfraService {
+
+ @Override
+ public void registerProperties() {
+ // NO-OP
+ }
+
+ @Override
+ public void initialize() {
+ registerProperties();
+ }
+
+ @Override
+ public void shutdown() {
+ // NO-OP
+ }
+
+ @Override
+ public String connectionString() {
+ return System.getProperty(AzureEventHubsProperties.CONNECTION_STRING);
+ }
+
+ @Override
+ public String blobEndpoint() {
+ return System.getProperty(AzureEventHubsProperties.BLOB_ENDPOINT,
+ String.format("https://%s.blob.core.windows.net";,
blobAccountName()));
+ }
+
+ @Override
+ public String blobAccountName() {
+ return System.getProperty(AzureEventHubsProperties.BLOB_ACCOUNT_NAME);
+ }
+
+ @Override
+ public String blobAccessKey() {
+ return System.getProperty(AzureEventHubsProperties.BLOB_ACCESS_KEY);
+ }
+}
diff --git
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsService.java
similarity index 57%
copy from
test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
copy to
test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsService.java
index ad5c47cd83da..69d9cd194910 100644
---
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsService.java
@@ -15,20 +15,13 @@
* limitations under the License.
*/
-package org.apache.camel.test.infra.azure.servicebus.common;
+package org.apache.camel.test.infra.azure.eventhubs.services;
-public final class AzureServiceBusProperties {
- /**
- * The connection string of the Service Bus namespace: the remote service
reads it, and the local service sets it to
- * the connection string of Floci AZ
- */
- public static final String CONNECTION_STRING =
"azure.servicebus.connection.string";
- /**
- * The Artemis image that Floci AZ starts as the Service Bus broker
- */
- public static final String BROKER_CONTAINER =
"azure.servicebus.broker.container";
+import org.apache.camel.test.infra.common.services.ContainerTestService;
+import org.apache.camel.test.infra.common.services.TestService;
- private AzureServiceBusProperties() {
-
- }
+/**
+ * Test infra service for Azure Event Hubs
+ */
+public interface AzureEventHubsService extends AzureEventHubsInfraService,
TestService, ContainerTestService {
}
diff --git
a/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsServiceFactory.java
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsServiceFactory.java
new file mode 100644
index 000000000000..2840f53fc81e
--- /dev/null
+++
b/test-infra/camel-test-infra-azure-eventhubs/src/main/java/org/apache/camel/test/infra/azure/eventhubs/services/AzureEventHubsServiceFactory.java
@@ -0,0 +1,92 @@
+/*
+ * 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.azure.eventhubs.services;
+
+import org.apache.camel.test.infra.common.services.SimpleTestServiceBuilder;
+import org.apache.camel.test.infra.common.services.SingletonService;
+
+/**
+ * Creates the Event Hubs service: the local one, unless the {@code
azure.instance.type} system property is
+ * {@code remote}.
+ */
+public final class AzureEventHubsServiceFactory {
+
+ private static class SingletonAzureEventHubsService extends
SingletonService<AzureEventHubsService>
+ implements AzureEventHubsService {
+ public SingletonAzureEventHubsService(AzureEventHubsService service,
String name) {
+ super(service, name);
+ }
+
+ @Override
+ public String connectionString() {
+ return getService().connectionString();
+ }
+
+ @Override
+ public String blobEndpoint() {
+ return getService().blobEndpoint();
+ }
+
+ @Override
+ public String blobAccountName() {
+ return getService().blobAccountName();
+ }
+
+ @Override
+ public String blobAccessKey() {
+ return getService().blobAccessKey();
+ }
+ }
+
+ private AzureEventHubsServiceFactory() {
+
+ }
+
+ public static SimpleTestServiceBuilder<AzureEventHubsService> builder() {
+ return new SimpleTestServiceBuilder<>("azure");
+ }
+
+ public static AzureEventHubsService createService() {
+ return builder()
+ .addLocalMapping(AzureEventHubsLocalContainerService::new)
+ .addRemoteMapping(AzureEventHubsRemoteService::new)
+ .build();
+ }
+
+ public static AzureEventHubsService createSingletonService() {
+ return SingletonServiceHolder.INSTANCE;
+ }
+
+ private static class SingletonServiceHolder {
+ static final AzureEventHubsService INSTANCE;
+ static {
+ SimpleTestServiceBuilder<AzureEventHubsService> instance =
builder();
+ instance.addLocalMapping(
+ () -> new SingletonAzureEventHubsService(new
AzureEventHubsLocalContainerService(), "azure-eventhubs"))
+ .addRemoteMapping(AzureEventHubsRemoteService::new);
+ INSTANCE = instance.build();
+ }
+ }
+
+ static class AzureEventHubsLocalContainerService extends
AzureEventHubsLocalContainerInfraService
+ implements AzureEventHubsService {
+ }
+
+ static class AzureEventHubsRemoteService extends
AzureEventHubsRemoteInfraService implements AzureEventHubsService {
+ }
+}
diff --git
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
b/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
index ad5c47cd83da..213b5be68a1a 100644
---
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
+++
b/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/common/AzureServiceBusProperties.java
@@ -23,10 +23,6 @@ public final class AzureServiceBusProperties {
* the connection string of Floci AZ
*/
public static final String CONNECTION_STRING =
"azure.servicebus.connection.string";
- /**
- * The Artemis image that Floci AZ starts as the Service Bus broker
- */
- public static final String BROKER_CONTAINER =
"azure.servicebus.broker.container";
private AzureServiceBusProperties() {
diff --git
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/services/AzureServiceBusLocalContainerInfraService.java
b/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/services/AzureServiceBusLocalContainerInfraService.java
index fca528eba436..c14c3e82d190 100644
---
a/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/services/AzureServiceBusLocalContainerInfraService.java
+++
b/test-infra/camel-test-infra-azure-servicebus/src/main/java/org/apache/camel/test/infra/azure/servicebus/services/AzureServiceBusLocalContainerInfraService.java
@@ -25,34 +25,18 @@ import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
import java.util.Objects;
-import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import com.github.dockerjava.api.DockerClient;
-import com.github.dockerjava.api.exception.NotFoundException;
-import com.github.dockerjava.api.model.Bind;
-import com.github.dockerjava.api.model.Container;
-import com.github.dockerjava.api.model.HostConfig;
-import com.github.dockerjava.api.model.Volume;
import org.apache.camel.spi.annotations.InfraService;
import org.apache.camel.test.infra.azure.common.services.FlociAzContainer;
import org.apache.camel.test.infra.azure.common.services.FlociAzInfraService;
import
org.apache.camel.test.infra.azure.servicebus.common.AzureServiceBusProperties;
-import org.apache.camel.test.infra.common.LocalPropertyResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.images.RemoteDockerImage;
import org.testcontainers.images.builder.Transferable;
-import org.testcontainers.utility.DockerImageName;
-import org.testcontainers.utility.ResourceReaper;
/**
* A local Service Bus namespace. Floci AZ serves the AMQP 1.0 data plane of
Service Bus with an Artemis broker, which
@@ -69,17 +53,13 @@ import org.testcontainers.utility.ResourceReaper;
public class AzureServiceBusLocalContainerInfraService extends
FlociAzInfraService implements AzureServiceBusInfraService {
private static final Logger LOG =
LoggerFactory.getLogger(AzureServiceBusLocalContainerInfraService.class);
- private static final String DOCKER_SOCKET = "/var/run/docker.sock";
// where the official Service Bus emulator reads its topology, which Floci
AZ reads as well
private static final String TOPOLOGY_FILE =
"/ServiceBus_Emulator/ConfigFiles/Config.json";
private static final String TOPOLOGY_RESOURCE = "Config.json";
// reports the namespaces and the host ports that Docker published their
broker on
private static final String NAMESPACES_PATH = "/" +
FlociAzContainer.DEFAULT_ACCOUNT_NAME + "-servicebus/namespaces";
private static final Pattern AMQP_HOST_PORT =
Pattern.compile("\"amqpHostPort\"\\s*:\\s*(\\d+)");
- // Floci AZ puts this namespace in the name of the containers it creates,
and in their floci_namespace label
- private static final String RESOURCE_NAMESPACE_ENV =
"FLOCI_AZ_DOCKER_RESOURCE_NAMESPACE";
- private static final String RESOURCE_NAMESPACE_LABEL = "floci_namespace";
- private static final String ARTEMIS_IMAGE_ENV =
"FLOCI_AZ_SERVICES_SERVICE_BUS_ARTEMIS_IMAGE";
+ private static final String ARTEMIS_IMAGE_VARIABLE =
"FLOCI_AZ_SERVICES_SERVICE_BUS_ARTEMIS_IMAGE";
private int amqpPort;
@@ -87,28 +67,13 @@ public class AzureServiceBusLocalContainerInfraService
extends FlociAzInfraServi
protected FlociAzContainer initContainer(String imageName) {
FlociAzContainer container = super.initContainer(imageName);
- container.withEnv("FLOCI_AZ_SERVICES_SERVICE_BUS_MOCKED", "false")
+ container.withDockerSocket()
+ .withEnv("FLOCI_AZ_SERVICES_SERVICE_BUS_MOCKED", "false")
// Docker publishes the broker on free host ports, read back
once it started, so that builds running at
// the same time on one Docker host do not clash
.withEnv("FLOCI_AZ_SERVICES_SERVICE_BUS_AMQP_PORT", "0")
.withEnv("FLOCI_AZ_SERVICES_SERVICE_BUS_AMQP_TLS_PORT", "0")
- // when it starts, Floci AZ removes the containers of its
namespace it finds, as left over: each
- // instance has a namespace of its own, so that it does not
remove the broker of another instance
- .withEnv(RESOURCE_NAMESPACE_ENV, "camel" +
UUID.randomUUID().toString().replace("-", "").substring(0, 12))
.withCopyToContainer(Transferable.of(readTopology()),
TOPOLOGY_FILE)
- .withCreateContainerCmdModifier(cmd -> {
- HostConfig hostConfig = cmd.getHostConfig();
- List<Bind> binds = new ArrayList<>();
- if (hostConfig.getBinds() != null) {
- binds.addAll(Arrays.asList(hostConfig.getBinds()));
- }
- binds.add(new Bind(
-
DockerClientFactory.instance().getRemoteDockerUnixSocketPath(), new
Volume(DOCKER_SOCKET)));
- hostConfig.withBinds(binds)
- // where SELinux is enforcing, it denies a
container the Docker socket of the host unless
- // the labeling of the container is disabled
- .withSecurityOpts(List.of("label=disable"));
- })
// Floci AZ answers once the broker started and the topology
is created
.waitingFor(Wait.forHttp(NAMESPACES_PATH)
.forPort(FlociAzContainer.SERVICE_PORT)
@@ -131,17 +96,19 @@ public class AzureServiceBusLocalContainerInfraService
extends FlociAzInfraServi
public void initialize() {
FlociAzContainer container = getContainer();
- // Floci AZ pulls the broker image before it answers: pull it here
instead, as Testcontainers pulls its images
- String brokerImage =
LocalPropertyResolver.getProperty(AzureServiceBusLocalContainerInfraService.class,
- AzureServiceBusProperties.BROKER_CONTAINER);
- container.withEnv(ARTEMIS_IMAGE_ENV, new
RemoteDockerImage(DockerImageName.parse(brokerImage)).get());
-
- // Testcontainers does not know the broker container, which Floci AZ
creates: have it removed when the JVM
- // exits, should the service not be shut down
-
ResourceReaper.instance().registerLabelsFilterForCleanup(Map.of(RESOURCE_NAMESPACE_LABEL,
resourceNamespace()));
-
- container.start();
- amqpPort = readAmqpPort();
+ try {
+ container.withBrokerImage(ARTEMIS_IMAGE_VARIABLE);
+ container.start();
+ amqpPort = readAmqpPort();
+ } catch (RuntimeException e) {
+ // also removes the broker, should Floci AZ have started it
+ try {
+ shutdown();
+ } catch (RuntimeException suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
LOG.info("Floci AZ running at address {}, with the Service Bus broker
at {}:{}", container.serviceAddress(), host(),
amqpPort);
@@ -153,8 +120,8 @@ public class AzureServiceBusLocalContainerInfraService
extends FlociAzInfraServi
HttpRequest request =
HttpRequest.newBuilder(URI.create(getContainer().serviceAddress() +
NAMESPACES_PATH))
.timeout(Duration.ofSeconds(30))
.build();
- try {
- String namespaces = HttpClient.newHttpClient().send(request,
HttpResponse.BodyHandlers.ofString()).body();
+ try (HttpClient client = HttpClient.newHttpClient()) {
+ String namespaces = client.send(request,
HttpResponse.BodyHandlers.ofString()).body();
Matcher matcher = AMQP_HOST_PORT.matcher(namespaces);
if (!matcher.find()) {
throw new IllegalStateException("Floci AZ reports no Service
Bus broker: " + namespaces);
@@ -173,36 +140,6 @@ public class AzureServiceBusLocalContainerInfraService
extends FlociAzInfraServi
System.setProperty(AzureServiceBusProperties.CONNECTION_STRING,
connectionString());
}
- @Override
- public void shutdown() {
- try {
- super.shutdown();
- } finally {
- removeBrokerContainers();
- }
- }
-
- // the broker containers are not removed with Floci AZ, which
Testcontainers kills
- private void removeBrokerContainers() {
- DockerClient client = DockerClientFactory.instance().client();
- List<Container> containers = client.listContainersCmd()
- .withShowAll(true)
- .withLabelFilter(Map.of(RESOURCE_NAMESPACE_LABEL,
resourceNamespace()))
- .exec();
- for (Container container : containers) {
- LOG.debug("Removing the container {} that Floci AZ started",
Arrays.toString(container.getNames()));
- try {
-
client.removeContainerCmd(container.getId()).withForce(true).withRemoveVolumes(true).exec();
- } catch (NotFoundException e) {
- LOG.debug("The container {} was already removed",
container.getId());
- }
- }
- }
-
- private String resourceNamespace() {
- return getContainer().getEnvMap().get(RESOURCE_NAMESPACE_ENV);
- }
-
@Override
public String connectionString() {
// Floci AZ ignores the key, and UseDevelopmentEmulator has the Azure
SDK connect with AMQP rather than AMQPS
diff --git
a/test-infra/camel-test-infra-azure-servicebus/src/main/resources/org/apache/camel/test/infra/azure/servicebus/services/container.properties
b/test-infra/camel-test-infra-azure-servicebus/src/main/resources/org/apache/camel/test/infra/azure/servicebus/services/container.properties
deleted file mode 100644
index 308e6b6203ec..000000000000
---
a/test-infra/camel-test-infra-azure-servicebus/src/main/resources/org/apache/camel/test/infra/azure/servicebus/services/container.properties
+++ /dev/null
@@ -1,22 +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.
-## ---------------------------------------------------------------------------
-# The Artemis broker that Floci AZ (azure.container) starts for the Service
Bus data plane. Floci AZ copies AMQP jars
-# patched for one Artemis version into it, so the version must be the one that
the Floci AZ image expects (its
-# FLOCI_AZ_SERVICES_SERVICE_BUS_ARTEMIS_IMAGE default): it is pinned, and
changes with azure.container. The key does
-# not end with "artemis": the test-infra metadata would take this version for
the Artemis services.
-azure.servicebus.broker.container=mirror.gcr.io/apache/activemq-artemis:2.44.0
-azure.servicebus.broker.container.version.include=2.44.0
diff --git a/test-infra/pom.xml b/test-infra/pom.xml
index d93566078221..95570efda9ef 100644
--- a/test-infra/pom.xml
+++ b/test-infra/pom.xml
@@ -52,6 +52,7 @@
<module>camel-test-infra-rabbitmq</module>
<module>camel-test-infra-azure-common</module>
<module>camel-test-infra-azure-eventgrid</module>
+ <module>camel-test-infra-azure-eventhubs</module>
<module>camel-test-infra-azure-functions</module>
<module>camel-test-infra-azure-key-vault</module>
<module>camel-test-infra-azure-servicebus</module>
