You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by bv...@apache.org on 2022/09/11 07:50:10 UTC
[camel] branch camel-3.18.x updated: CAMEL-18451: Add Azure-AD authentication support for camel-azure-eventhubs
This is an automated email from the ASF dual-hosted git repository.
bvahdat pushed a commit to branch camel-3.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-3.18.x by this push:
new 2ce8f508dde CAMEL-18451: Add Azure-AD authentication support for camel-azure-eventhubs
2ce8f508dde is described below
commit 2ce8f508ddee4866e6fcf158739dd3ab005856da
Author: Babak Vahdat <bv...@apache.org>
AuthorDate: Sat Sep 10 17:57:14 2022 +0200
CAMEL-18451: Add Azure-AD authentication support for camel-azure-eventhubs
---
camel-dependencies/pom.xml | 20 +++----
.../camel-azure/camel-azure-eventhubs/pom.xml | 5 ++
.../eventhubs/EventHubsComponentConfigurer.java | 6 ++
.../eventhubs/EventHubsEndpointConfigurer.java | 6 ++
.../eventhubs/EventHubsEndpointUriFactory.java | 6 +-
.../services/org/apache/camel/component.properties | 2 +-
.../component/azure/eventhubs/azure-eventhubs.json | 36 ++++++------
.../src/main/docs/azure-eventhubs-component.adoc | 22 +++++++-
.../azure/eventhubs/EventHubsComponent.java | 62 ++++++++++++++++-----
.../azure/eventhubs/EventHubsConfiguration.java | 38 ++++++++-----
.../azure/eventhubs/EventHubsConsumer.java | 1 +
.../azure/eventhubs/EventHubsProducer.java | 1 +
.../eventhubs/client/EventHubsClientFactory.java | 65 ++++++++++++++++++----
.../azure/eventhubs/EventHubsComponentTest.java | 29 +++++++---
.../azure/eventhubs/EventProcessorTest.java | 29 +++++++++-
parent/pom.xml | 20 +++----
.../azure/common/services/AzuriteContainer.java | 2 +-
17 files changed, 256 insertions(+), 94 deletions(-)
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index a4299913340..2c86d387fab 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -65,17 +65,17 @@
<aws-java-sdk2-version>2.17.250</aws-java-sdk2-version>
<aws-xray-version>2.6.1</aws-xray-version>
<axiom-version>1.2.14</axiom-version>
- <azure-cosmos-version>4.29.1</azure-cosmos-version>
- <azure-eventhubs-version>5.11.2</azure-eventhubs-version>
- <azure-identity-version>1.5.1</azure-identity-version>
+ <azure-cosmos-version>4.35.0</azure-cosmos-version>
+ <azure-eventhubs-version>5.13.0</azure-eventhubs-version>
+ <azure-identity-version>1.5.4</azure-identity-version>
<azure-jackson2-version>2.13.3</azure-jackson2-version>
- <azure-keyvault-secrets-version>4.4.2</azure-keyvault-secrets-version>
- <azure-messaging-eventhubs-checkpointstore-blob-version>1.12.1</azure-messaging-eventhubs-checkpointstore-blob-version>
- <azure-messaging-servicebus-version>7.8.0</azure-messaging-servicebus-version>
- <azure-storage-blob-changefeed-version>12.0.0-beta.17</azure-storage-blob-changefeed-version>
- <azure-storage-blob-version>12.16.1</azure-storage-blob-version>
- <azure-storage-datalake-version>12.9.1</azure-storage-datalake-version>
- <azure-storage-queue-version>12.12.2</azure-storage-queue-version>
+ <azure-keyvault-secrets-version>4.4.6</azure-keyvault-secrets-version>
+ <azure-messaging-eventhubs-checkpointstore-blob-version>1.15.0</azure-messaging-eventhubs-checkpointstore-blob-version>
+ <azure-messaging-servicebus-version>7.10.1</azure-messaging-servicebus-version>
+ <azure-storage-blob-changefeed-version>12.0.0-beta.18</azure-storage-blob-changefeed-version>
+ <azure-storage-blob-version>12.19.0</azure-storage-blob-version>
+ <azure-storage-datalake-version>12.12.0</azure-storage-datalake-version>
+ <azure-storage-queue-version>12.14.1</azure-storage-queue-version>
<bouncycastle-version>1.71</bouncycastle-version>
<box-java-sdk-version>2.58.0</box-java-sdk-version>
<braintree-gateway-version>3.16.1</braintree-gateway-version>
diff --git a/components/camel-azure/camel-azure-eventhubs/pom.xml b/components/camel-azure/camel-azure-eventhubs/pom.xml
index b51d5f955f0..fd269a51e06 100644
--- a/components/camel-azure/camel-azure-eventhubs/pom.xml
+++ b/components/camel-azure/camel-azure-eventhubs/pom.xml
@@ -58,6 +58,11 @@
<artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
<version>${azure-messaging-eventhubs-checkpointstore-blob-version}</version>
</dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ <version>${azure-identity-version}</version>
+ </dependency>
<!-- extras -->
<dependency>
diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java
index 43e0a8e8dd4..89376dba808 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java
@@ -67,6 +67,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl
case "sharedAccessKey": getOrCreateConfiguration(target).setSharedAccessKey(property(camelContext, java.lang.String.class, value)); return true;
case "sharedaccessname":
case "sharedAccessName": getOrCreateConfiguration(target).setSharedAccessName(property(camelContext, java.lang.String.class, value)); return true;
+ case "tokencredential":
+ case "tokenCredential": getOrCreateConfiguration(target).setTokenCredential(property(camelContext, com.azure.core.credential.TokenCredential.class, value)); return true;
default: return false;
}
}
@@ -118,6 +120,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl
case "sharedAccessKey": return java.lang.String.class;
case "sharedaccessname":
case "sharedAccessName": return java.lang.String.class;
+ case "tokencredential":
+ case "tokenCredential": return com.azure.core.credential.TokenCredential.class;
default: return null;
}
}
@@ -165,6 +169,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl
case "sharedAccessKey": return getOrCreateConfiguration(target).getSharedAccessKey();
case "sharedaccessname":
case "sharedAccessName": return getOrCreateConfiguration(target).getSharedAccessName();
+ case "tokencredential":
+ case "tokenCredential": return getOrCreateConfiguration(target).getTokenCredential();
default: return null;
}
}
diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java
index 256cd00f9a1..f94b3af0678 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java
@@ -61,6 +61,8 @@ public class EventHubsEndpointConfigurer extends PropertyConfigurerSupport imple
case "sharedAccessKey": target.getConfiguration().setSharedAccessKey(property(camelContext, java.lang.String.class, value)); return true;
case "sharedaccessname":
case "sharedAccessName": target.getConfiguration().setSharedAccessName(property(camelContext, java.lang.String.class, value)); return true;
+ case "tokencredential":
+ case "tokenCredential": target.getConfiguration().setTokenCredential(property(camelContext, com.azure.core.credential.TokenCredential.class, value)); return true;
default: return false;
}
}
@@ -113,6 +115,8 @@ public class EventHubsEndpointConfigurer extends PropertyConfigurerSupport imple
case "sharedAccessKey": return java.lang.String.class;
case "sharedaccessname":
case "sharedAccessName": return java.lang.String.class;
+ case "tokencredential":
+ case "tokenCredential": return com.azure.core.credential.TokenCredential.class;
default: return null;
}
}
@@ -161,6 +165,8 @@ public class EventHubsEndpointConfigurer extends PropertyConfigurerSupport imple
case "sharedAccessKey": return target.getConfiguration().getSharedAccessKey();
case "sharedaccessname":
case "sharedAccessName": return target.getConfiguration().getSharedAccessName();
+ case "tokencredential":
+ case "tokenCredential": return target.getConfiguration().getTokenCredential();
default: return null;
}
}
diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointUriFactory.java b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointUriFactory.java
index a67cc23bb2f..63d3afb6e9a 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointUriFactory.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointUriFactory.java
@@ -21,7 +21,7 @@ public class EventHubsEndpointUriFactory extends org.apache.camel.support.compon
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(22);
+ Set<String> props = new HashSet<>(23);
props.add("amqpRetryOptions");
props.add("amqpTransportType");
props.add("blobAccessKey");
@@ -44,12 +44,14 @@ public class EventHubsEndpointUriFactory extends org.apache.camel.support.compon
props.add("producerAsyncClient");
props.add("sharedAccessKey");
props.add("sharedAccessName");
+ props.add("tokenCredential");
PROPERTY_NAMES = Collections.unmodifiableSet(props);
- Set<String> secretProps = new HashSet<>(4);
+ Set<String> secretProps = new HashSet<>(5);
secretProps.add("blobAccessKey");
secretProps.add("blobStorageSharedKeyCredential");
secretProps.add("connectionString");
secretProps.add("sharedAccessKey");
+ secretProps.add("tokenCredential");
SECRET_PROPERTY_NAMES = Collections.unmodifiableSet(secretProps);
MULTI_VALUE_PREFIXES = Collections.emptySet();
}
diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties
index 0043292eeaa..3b8cff699ad 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties
+++ b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties
@@ -2,6 +2,6 @@
components=azure-eventhubs
groupId=org.apache.camel
artifactId=camel-azure-eventhubs
-version=3.18.2-SNAPSHOT
+version=3.18.3-SNAPSHOT
projectName=Camel :: Azure :: Event Hubs
projectDescription=Camel Azure Event Hubs Component
diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
index 78d30b90560..71ad7644386 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
+++ b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
@@ -11,7 +11,7 @@
"supportLevel": "Stable",
"groupId": "org.apache.camel",
"artifactId": "camel-azure-eventhubs",
- "version": "3.18.2-SNAPSHOT",
+ "version": "3.18.3-SNAPSHOT",
"scheme": "azure-eventhubs",
"extendsScheme": "",
"syntax": "azure-eventhubs:namespace\/eventHubName",
@@ -25,13 +25,13 @@
"amqpRetryOptions": { "kind": "property", "displayName": "Amqp Retry Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the retry policy for EventHubAsyncClient. If not specified, the default r [...]
"amqpTransportType": { "kind": "property", "displayName": "Amqp Transport Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpTransportType", "enum": [ "Amqp", "AmqpWebSockets" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "AMQP", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the tran [...]
"configuration": { "kind": "property", "displayName": "Configuration", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "deprecated": false, "autowired": false, "secret": false, "description": "The component configurations" },
- "blobAccessKey": { "kind": "property", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure acco [...]
+ "blobAccessKey": { "kind": "property", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure acco [...]
"blobAccountName": { "kind": "property", "displayName": "Blob Account Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for a [...]
- "blobContainerName": { "kind": "property", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall b [...]
- "blobStorageSharedKeyCredential": { "kind": "property", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default [...]
+ "blobContainerName": { "kind": "property", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall b [...]
+ "blobStorageSharedKeyCredential": { "kind": "property", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default [...]
"bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a me [...]
- "checkpointStore": { "kind": "property", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the [...]
- "consumerGroupName": { "kind": "property", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated [...]
+ "checkpointStore": { "kind": "property", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the [...]
+ "consumerGroupName": { "kind": "property", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated [...]
"eventPosition": { "kind": "property", "displayName": "Event Position", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, com.azure.messaging.eventhubs.models.EventPosition>", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the map containing the event [...]
"prefetchCount": { "kind": "property", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the count used by the receiver to control the number of events the Event Hub consumer w [...]
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
@@ -40,8 +40,9 @@
"producerAsyncClient": { "kind": "property", "displayName": "Producer Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.EventHubProducerAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the EventHubProducer [...]
"autowiredEnabled": { "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which t [...]
"connectionString": { "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply t [...]
- "sharedAccessKey": { "kind": "property", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName" },
- "sharedAccessName": { "kind": "property", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys" }
+ "sharedAccessKey": { "kind": "property", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName." },
+ "sharedAccessName": { "kind": "property", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys." },
+ "tokenCredential": { "kind": "property", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Still another way of authentication (beside supplying namespace, sharedAc [...]
},
"headers": {
"CamelAzureEventHubsPartitionKey": { "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "(producer) Overrides the hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a specific partition. (consumer) It sets the partition hashing key if it was set when originally publishing t [...]
@@ -53,16 +54,16 @@
"CamelMessageTimestamp": { "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The timestamp of the message", "constantName": "org.apache.camel.component.azure.eventhubs.EventHubsConstants#MESSAGE_TIMESTAMP" }
},
"properties": {
- "namespace": { "kind": "path", "displayName": "Namespace", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs namespace created in Azure Portal" },
- "eventHubName": { "kind": "path", "displayName": "Event Hub Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs name under a specific namcespace" },
+ "namespace": { "kind": "path", "displayName": "Namespace", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs namespace created in Azure Portal." },
+ "eventHubName": { "kind": "path", "displayName": "Event Hub Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs name under a specific namespace." },
"amqpRetryOptions": { "kind": "parameter", "displayName": "Amqp Retry Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the retry policy for EventHubAsyncClient. If not specified, the default [...]
"amqpTransportType": { "kind": "parameter", "displayName": "Amqp Transport Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpTransportType", "enum": [ "Amqp", "AmqpWebSockets" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "AMQP", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the tra [...]
- "blobAccessKey": { "kind": "parameter", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure acc [...]
+ "blobAccessKey": { "kind": "parameter", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure acc [...]
"blobAccountName": { "kind": "parameter", "displayName": "Blob Account Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for [...]
- "blobContainerName": { "kind": "parameter", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall [...]
- "blobStorageSharedKeyCredential": { "kind": "parameter", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default [...]
- "checkpointStore": { "kind": "parameter", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the [...]
- "consumerGroupName": { "kind": "parameter", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated [...]
+ "blobContainerName": { "kind": "parameter", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall [...]
+ "blobStorageSharedKeyCredential": { "kind": "parameter", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default [...]
+ "checkpointStore": { "kind": "parameter", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the [...]
+ "consumerGroupName": { "kind": "parameter", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated [...]
"eventPosition": { "kind": "parameter", "displayName": "Event Position", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, com.azure.messaging.eventhubs.models.EventPosition>", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the map containing the event [...]
"prefetchCount": { "kind": "parameter", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the count used by the receiver to control the number of events the Event Hub consumer [...]
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now [...]
@@ -73,7 +74,8 @@
"producerAsyncClient": { "kind": "parameter", "displayName": "Producer Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.EventHubProducerAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the EventHubProduce [...]
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may other [...]
"connectionString": { "kind": "parameter", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply [...]
- "sharedAccessKey": { "kind": "parameter", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName" },
- "sharedAccessName": { "kind": "parameter", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys" }
+ "sharedAccessKey": { "kind": "parameter", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName." },
+ "sharedAccessName": { "kind": "parameter", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys." },
+ "tokenCredential": { "kind": "parameter", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Still another way of authentication (beside supplying namespace, sharedA [...]
}
}
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 b958e2ef31e..0b0b667287b 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
@@ -72,8 +72,9 @@ be generated through your Event Hubs Azure portal.
- Provide `connectionString` string, if you provide the connection string, you don't supply `namespace`, `eventHubName`, `sharedAccessKey` and `sharedAccessName`
as these data already included in the `connectionString`, therefore is the simplest option to get started. Learn more https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string[here] on how to generate the connection string.
- Provide a https://docs.microsoft.com/en-us/java/api/com.azure.messaging.eventhubs.eventhubproducerasyncclient?view=azure-java-stable[EventHubProducerAsyncClient] instance which can be
-provided into `producerAsyncClient`. However, this is *only possible for camel producer*, for the camel consumer, is not possible to inject the client due to some design constrain by the `EventProcessorClient`.
-
+provided into `producerAsyncClient`. However, this is *only possible for camel producer*, for the camel consumer, is not possible to inject the client due to some design constraint by the `EventProcessorClient`.
+- Provide an implementation of `com.azure.core.credential.TokenCredential` into the Camel's Registry, e.g. using the `com.azure.identity.DefaultAzureCredentialBuilder().build();` API.
+See the documentation https://docs.microsoft.com/en-us/azure/active-directory/authentication/overview-authentication[here about Azure-AD authentication].
== Checkpoint Store Information
A checkpoint store stores and retrieves partition ownership information and checkpoint details for each partition in a given consumer group of an event hub instance. Users are not meant to implement an CheckpointStore.
@@ -160,6 +161,21 @@ from("direct:start")
.to("azure-eventhubs:?connectionString=RAW({{connectionString}})"
----
+=== Azure-AD Authentication example
+The example below makes use of the Azure-AD authentication. See https://docs.microsoft.com/en-us/java/api/overview/azure/identity-readme?view=azure-java-stable#environment-variables[here]
+about what environment variables you need to set for this to work:
+
+[source,java]
+----
+@BindToRegistry("myTokenCredential")
+public com.azure.core.credential.TokenCredential myTokenCredential() {
+ return com.azure.identity.DefaultAzureCredentialBuilder().build();
+}
+
+from("direct:start")
+.to("azure-eventhubs:namespace/eventHubName?tokenCredential=#myTokenCredential)"
+----
+
=== Development Notes (Important)
When developing on this component, you will need to obtain your Azure accessKey in order to run the integration tests. In addition to the mocked unit tests
@@ -168,7 +184,7 @@ To run the integration tests, on this component directory, run the following mav
[source,bash]
----
-mvn verify -PfullTests -DconnectionString=string -DblobAccountName=blob -DblobAccessKey=key
+mvn verify -DconnectionString=string -DblobAccountName=blob -DblobAccessKey=key
----
Whereby `blobAccountName` is your Azure account name and `blobAccessKey` is the access key being generated from Azure portal and `connectionString` is the eventHub connection string.
diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java
index 169e0be9049..4f672d0b339 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java
@@ -17,7 +17,10 @@
package org.apache.camel.component.azure.eventhubs;
import java.util.Map;
+import java.util.Set;
+import com.azure.core.credential.TokenCredential;
+import com.azure.identity.DefaultAzureCredentialBuilder;
import org.apache.camel.Endpoint;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.annotations.Component;
@@ -42,20 +45,23 @@ public class EventHubsComponent extends DefaultComponent {
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
-
- final EventHubsConfiguration configuration
- = this.configuration != null ? this.configuration.copy() : new EventHubsConfiguration();
+ final EventHubsConfiguration configuration = this.configuration.copy();
final EventHubsEndpoint endpoint = new EventHubsEndpoint(uri, this, configuration);
setProperties(endpoint, parameters);
- // if we don't have client nor connectionString, we check for params
- if (areAzureClientsNotSet(configuration) && ObjectHelper.isEmpty(configuration.getConnectionString())
- && ObjectHelper.isEmpty(configuration.getProducerAsyncClient())) {
+ // if TokenCredential is set or we don't have client nor connectionString, we then check and use remaining
+ if (isTokenCredentialSet(configuration)
+ || !isProducerAsyncClientSet(configuration) && !isConnectionStringSet(configuration)) {
checkAndSetNamespaceAndHubName(configuration, remaining);
- validateConfigurations(configuration);
}
+ if (isTokenCredentialSet(configuration)) {
+ setTokenCredential(configuration);
+ }
+
+ validateConfigurations(configuration);
+
return endpoint;
}
@@ -71,10 +77,10 @@ public class EventHubsComponent extends DefaultComponent {
}
private void validateConfigurations(final EventHubsConfiguration configuration) {
- if (!isAccessKeyAndAccessNameSet(configuration)) {
+ if (!isAccessKeyAndAccessNameSet(configuration) && !isProducerAsyncClientSet(configuration)
+ && !isConnectionStringSet(configuration) && !isTokenCredentialSet(configuration)) {
throw new IllegalArgumentException(
- "Azure EventHubs SharedAccessName/SharedAccessKey, ConsumerAsyncClient/ProducerAsyncClient "
- + "or connectionString must be specified.");
+ "Azure EventHubs SharedAccessName/SharedAccessKey, ProducerAsyncClient, ConnectionString or TokenCredential must be specified.");
}
}
@@ -83,22 +89,48 @@ public class EventHubsComponent extends DefaultComponent {
&& ObjectHelper.isNotEmpty(configuration.getSharedAccessKey());
}
- private boolean areAzureClientsNotSet(final EventHubsConfiguration configuration) {
- return ObjectHelper.isEmpty(configuration.getProducerAsyncClient());
+ private boolean isConnectionStringSet(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isNotEmpty(configuration.getConnectionString());
+ }
+
+ private boolean isTokenCredentialSet(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isNotEmpty(configuration.getTokenCredential());
+ }
+
+ private boolean isProducerAsyncClientSet(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isNotEmpty(configuration.getProducerAsyncClient());
}
private void checkAndSetNamespaceAndHubName(final EventHubsConfiguration configuration, final String remaining) {
// only set if clients are empty and remaining exists
+ final String errorMessage = "ConnectionString, ProducerAsyncClient or Namespace and EventHub name must be set";
if (ObjectHelper.isEmpty(remaining)) {
- throw new IllegalArgumentException("ConnectionString, AzureClients or Namespace and EventHub name must be set");
+ throw new IllegalArgumentException(errorMessage);
}
final String[] parts = remaining.split("/");
- if (parts.length < 2) {
- throw new IllegalArgumentException("ConnectionString, AzureClients or Namespace and EventHub name must be set");
+ if (parts.length != 2) {
+ throw new IllegalArgumentException(errorMessage);
}
configuration.setNamespace(parts[0]);
configuration.setEventHubName(parts[1]);
}
+
+ private void setTokenCredential(final EventHubsConfiguration configuration) {
+ final Set<TokenCredential> tokenCredentialFromRegistry
+ = getCamelContext().getRegistry().findByType(TokenCredential.class);
+
+ // find exactly one from the registry or create one
+ if (tokenCredentialFromRegistry.size() == 1) {
+ final TokenCredential tokenCredential = tokenCredentialFromRegistry.stream().findFirst().get();
+ configuration.setTokenCredential(tokenCredential);
+ LOG.debug("using the provided TokenCredential instance: {} for the Azure-AD authentication", tokenCredential);
+ } else {
+ final TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();
+ configuration.setTokenCredential(tokenCredential);
+ LOG.debug("using the DefaultAzureCredential instance: {} for the Azure-AD authentication", tokenCredential);
+ }
+ }
+
}
diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java
index 6a4b24fdbf0..c59ee200177 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java
@@ -21,6 +21,7 @@ import java.util.Map;
import com.azure.core.amqp.AmqpRetryOptions;
import com.azure.core.amqp.AmqpTransportType;
+import com.azure.core.credential.TokenCredential;
import com.azure.messaging.eventhubs.CheckpointStore;
import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
@@ -46,6 +47,8 @@ public class EventHubsConfiguration implements Cloneable {
private String sharedAccessKey;
@UriParam(label = "security", secret = true)
private String connectionString;
+ @UriParam(label = "security", secret = true)
+ private TokenCredential tokenCredential;
@UriParam(label = "common", defaultValue = "AMQP")
private AmqpTransportType amqpTransportType = AmqpTransportType.AMQP;
@UriParam(label = "common")
@@ -75,7 +78,7 @@ public class EventHubsConfiguration implements Cloneable {
private String partitionId;
/**
- * EventHubs namespace created in Azure Portal
+ * EventHubs namespace created in Azure Portal.
*/
public String getNamespace() {
return namespace;
@@ -86,7 +89,7 @@ public class EventHubsConfiguration implements Cloneable {
}
/**
- * EventHubs name under a specific namcespace
+ * EventHubs name under a specific namespace.
*/
public String getEventHubName() {
return eventHubName;
@@ -97,7 +100,7 @@ public class EventHubsConfiguration implements Cloneable {
}
/**
- * The name you chose for your EventHubs SAS keys
+ * The name you chose for your EventHubs SAS keys.
*/
public String getSharedAccessName() {
return sharedAccessName;
@@ -108,7 +111,7 @@ public class EventHubsConfiguration implements Cloneable {
}
/**
- * The generated value for the SharedAccessName
+ * The generated value for the SharedAccessName.
*/
public String getSharedAccessKey() {
return sharedAccessKey;
@@ -132,6 +135,18 @@ public class EventHubsConfiguration implements Cloneable {
this.connectionString = connectionString;
}
+ /**
+ * Still another way of authentication (beside supplying namespace, sharedAccessKey, sharedAccessName or connection
+ * string) is through Azure-AD authentication using an implementation instance of {@link TokenCredential}.
+ */
+ public TokenCredential getTokenCredential() {
+ return tokenCredential;
+ }
+
+ public void setTokenCredential(TokenCredential tokenCredential) {
+ this.tokenCredential = tokenCredential;
+ }
+
/**
* Sets the transport type by which all the communication with Azure Event Hubs occurs. Default value is
* {@link AmqpTransportType#AMQP}.
@@ -157,8 +172,7 @@ public class EventHubsConfiguration implements Cloneable {
/**
* Sets the name of the consumer group this consumer is associated with. Events are read in the context of this
- * group. The name of the consumer group that is created by default is {@link #DEFAULT_CONSUMER_GROUP_NAME
- * "$Default"}.
+ * group. The name of the consumer group that is created by default is {@code "$Default"}.
*/
public String getConsumerGroupName() {
return consumerGroupName;
@@ -239,7 +253,7 @@ public class EventHubsConfiguration implements Cloneable {
* </p>
*
* By default it set to use {@link com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore} which
- * stores all checkpoint offsets into Azure Blob Storage
+ * stores all checkpoint offsets into Azure Blob Storage.
*/
public CheckpointStore getCheckpointStore() {
return checkpointStore;
@@ -263,7 +277,7 @@ public class EventHubsConfiguration implements Cloneable {
/**
* In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to
- * be used for authentication with azure blob services
+ * be used for authentication with azure blob services.
*/
public String getBlobAccessKey() {
return blobAccessKey;
@@ -275,7 +289,7 @@ public class EventHubsConfiguration implements Cloneable {
/**
* In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the
- * BlobCheckpointStore to store the checkpoint offsets
+ * BlobCheckpointStore to store the checkpoint offsets.
*/
public String getBlobContainerName() {
return blobContainerName;
@@ -287,7 +301,7 @@ public class EventHubsConfiguration implements Cloneable {
/**
* In case you chose the default BlobCheckpointStore, StorageSharedKeyCredential can be injected to create the azure
- * client, this holds the important authentication information
+ * client, this holds the important authentication information.
*/
public StorageSharedKeyCredential getBlobStorageSharedKeyCredential() {
return blobStorageSharedKeyCredential;
@@ -311,10 +325,6 @@ public class EventHubsConfiguration implements Cloneable {
this.eventPosition = eventPosition;
}
- // *************************************************
- //
- // *************************************************
-
public EventHubsConfiguration copy() {
try {
return (EventHubsConfiguration) super.clone();
diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
index ef3230383b1..ddc73161768 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
@@ -58,6 +58,7 @@ public class EventHubsConsumer extends DefaultConsumer {
if (processorClient != null) {
// shutdown the client
processorClient.stop();
+ processorClient = null;
}
// shutdown camel consumer
diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java
index 1164156f8b4..b91afa7bfcb 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java
@@ -61,6 +61,7 @@ public class EventHubsProducer extends DefaultAsyncProducer {
if (producerAsyncClient != null) {
// shutdown async client
producerAsyncClient.close();
+ producerAsyncClient = null;
}
super.doStop();
diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java
index cb1a2ce84aa..a448f113040 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java
@@ -43,37 +43,69 @@ public final class EventHubsClientFactory {
}
public static EventHubProducerAsyncClient createEventHubProducerAsyncClient(final EventHubsConfiguration configuration) {
- return new EventHubClientBuilder()
- .connectionString(buildConnectionString(configuration))
+ EventHubClientBuilder eventHubClientBuilder = new EventHubClientBuilder()
.transportType(configuration.getAmqpTransportType())
- .retryOptions(configuration.getAmqpRetryOptions())
+ .retryOptions(configuration.getAmqpRetryOptions());
+
+ if (ObjectHelper.isEmpty(configuration.getTokenCredential())) {
+ return eventHubClientBuilder
+ .connectionString(buildConnectionString(configuration))
+ .buildAsyncProducerClient();
+ }
+
+ checkTokenCredentialConfiguration(configuration);
+ return eventHubClientBuilder
+ .fullyQualifiedNamespace(getFullyQualifiedNamespace(configuration))
+ .eventHubName(configuration.getEventHubName())
+ .credential(configuration.getTokenCredential())
.buildAsyncProducerClient();
}
public static EventHubConsumerAsyncClient createEventHubConsumerAsyncClient(final EventHubsConfiguration configuration) {
- return new EventHubClientBuilder()
- .connectionString(buildConnectionString(configuration))
+ EventHubClientBuilder eventHubClientBuilder = new EventHubClientBuilder()
.consumerGroup(configuration.getConsumerGroupName())
.prefetchCount(configuration.getPrefetchCount())
.transportType(configuration.getAmqpTransportType())
- .retryOptions(configuration.getAmqpRetryOptions())
+ .retryOptions(configuration.getAmqpRetryOptions());
+
+ if (ObjectHelper.isEmpty(configuration.getTokenCredential())) {
+ return eventHubClientBuilder
+ .connectionString(buildConnectionString(configuration))
+ .buildAsyncConsumerClient();
+ }
+
+ checkTokenCredentialConfiguration(configuration);
+ return eventHubClientBuilder
+ .fullyQualifiedNamespace(getFullyQualifiedNamespace(configuration))
+ .eventHubName(configuration.getEventHubName())
+ .credential(configuration.getTokenCredential())
.buildAsyncConsumerClient();
}
public static EventProcessorClient createEventProcessorClient(
final EventHubsConfiguration configuration, final Consumer<EventContext> processEvent,
final Consumer<ErrorContext> processError) {
- return new EventProcessorClientBuilder()
+ EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
.initialPartitionEventPosition(configuration.getEventPosition())
- .connectionString(buildConnectionString(configuration))
.checkpointStore(createCheckpointStore(configuration))
.consumerGroup(configuration.getConsumerGroupName())
.retryOptions(configuration.getAmqpRetryOptions())
.transportType(configuration.getAmqpTransportType())
.processError(processError)
- .processEvent(processEvent)
- .buildEventProcessorClient();
+ .processEvent(processEvent);
+ if (ObjectHelper.isEmpty(configuration.getTokenCredential())) {
+ return eventProcessorClientBuilder
+ .connectionString(buildConnectionString(configuration))
+ .buildEventProcessorClient();
+ }
+
+ checkTokenCredentialConfiguration(configuration);
+ return eventProcessorClientBuilder
+ .fullyQualifiedNamespace(getFullyQualifiedNamespace(configuration))
+ .eventHubName(configuration.getEventHubName())
+ .credential(configuration.getTokenCredential())
+ .buildEventProcessorClient();
}
// public for testing purposes
@@ -85,14 +117,19 @@ public final class EventHubsClientFactory {
.buildAsyncClient();
}
+ private static void checkTokenCredentialConfiguration(final EventHubsConfiguration configuration) {
+ if (ObjectHelper.isEmpty(configuration.getNamespace()) || ObjectHelper.isEmpty(configuration.getEventHubName())) {
+ throw new IllegalArgumentException("EventHub's namespace and name is required for the Azure-AD authentication");
+ }
+ }
+
private static CheckpointStore createCheckpointStore(final EventHubsConfiguration configuration) {
if (ObjectHelper.isNotEmpty(configuration.getCheckpointStore())) {
return configuration.getCheckpointStore();
}
// so we have no checkpoint store, we fallback to default BlobCheckpointStore
// first we check if we have all required params for BlobCheckpointStore
- if (ObjectHelper.isEmpty(configuration.getBlobContainerName())
- || !isCredentialsSet(configuration)) {
+ if (ObjectHelper.isEmpty(configuration.getBlobContainerName()) || !isCredentialsSet(configuration)) {
throw new IllegalArgumentException(
"Since there is no provided CheckpointStore, you will need to set blobAccountName, blobAccessName"
+ " or blobContainerName in order to use the default BlobCheckpointStore");
@@ -141,4 +178,8 @@ public final class EventHubsClientFactory {
? configuration.getBlobStorageSharedKeyCredential().getAccountName()
: configuration.getBlobAccountName();
}
+
+ private static String getFullyQualifiedNamespace(EventHubsConfiguration configuration) {
+ return configuration.getNamespace() + "." + SERVICE_URI_SEGMENT;
+ }
}
diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java
index 1f6b7f9124e..6f5fcad6a52 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.azure.eventhubs;
+import com.azure.identity.DefaultAzureCredential;
+import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
import org.apache.camel.ResolveEndpointFailedException;
@@ -25,6 +27,7 @@ import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -36,20 +39,21 @@ class EventHubsComponentTest extends CamelTestSupport {
() -> context.getEndpoint("azure-eventhubs:?sharedAccessKey=string&sharedAccessName=name"));
assertTrue(
- exception.getMessage().contains("ConnectionString, AzureClients or Namespace and EventHub name must be set"));
+ exception.getMessage()
+ .contains("ConnectionString, ProducerAsyncClient or Namespace and EventHub name must be set"));
ResolveEndpointFailedException exception2 = assertThrows(ResolveEndpointFailedException.class,
() -> context.getEndpoint("azure-eventhubs:name?sharedAccessKey=string&sharedAccessName=name"));
assertTrue(
- exception2.getMessage().contains("ConnectionString, AzureClients or Namespace and EventHub name must be set"));
+ exception2.getMessage()
+ .contains("ConnectionString, ProducerAsyncClient or Namespace and EventHub name must be set"));
}
@Test
- public void testCreateEndpointWithNoSuppliedClientsOrKeysOrConnectionString() {
+ public void testCreateEndpointWithNoSuppliedClientsOrKeysOrConnectionStringOrTokenCredential() {
final String expectedErrorMessage
- = "Azure EventHubs SharedAccessName/SharedAccessKey, ConsumerAsyncClient/ProducerAsyncClient "
- + "or connectionString must be specified";
+ = "Azure EventHubs SharedAccessName/SharedAccessKey, ProducerAsyncClient, ConnectionString or TokenCredential must be specified";
// first case: with no client or key or connectionstring
assertTrue(getErrorMessage("azure-eventhubs:name/hubName?").contains(expectedErrorMessage));
@@ -102,9 +106,18 @@ class EventHubsComponentTest extends CamelTestSupport {
context.getRegistry().bind("producerClient", producerAsyncClient);
final EventHubsEndpoint endpoint
- = (EventHubsEndpoint) context.getEndpoint("azure-eventhubs:name/hubName?producerAsyncClient=#producerClient");
+ = context.getEndpoint("azure-eventhubs:name/hubName?producerAsyncClient=#producerClient",
+ EventHubsEndpoint.class);
- assertEquals(producerAsyncClient, endpoint.getConfiguration().getProducerAsyncClient());
+ assertSame(producerAsyncClient, endpoint.getConfiguration().getProducerAsyncClient());
+
+ final DefaultAzureCredential tokenCredential = new DefaultAzureCredentialBuilder().build();
+ context.getRegistry().bind("tokenCredential", tokenCredential);
+
+ final EventHubsEndpoint endpoint2
+ = context.getEndpoint("azure-eventhubs:name/hubName?tokenCredential=#tokenCredential", EventHubsEndpoint.class);
+
+ assertSame(tokenCredential, endpoint2.getConfiguration().getTokenCredential());
}
@Test
@@ -113,7 +126,7 @@ class EventHubsComponentTest extends CamelTestSupport {
+ "&sharedAccessKey=DummyKey"
+ "&consumerGroupName=testConsumer&prefetchCount=100";
- final EventHubsEndpoint endpoint = (EventHubsEndpoint) context.getEndpoint(uri);
+ final EventHubsEndpoint endpoint = context.getEndpoint(uri, EventHubsEndpoint.class);
assertEquals("namespace", endpoint.getConfiguration().getNamespace());
assertEquals("hubName", endpoint.getConfiguration().getEventHubName());
diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java
index 9b4ba1fa6ee..03c875d2325 100644
--- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java
+++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.azure.eventhubs;
import java.util.function.Consumer;
+import com.azure.identity.IntelliJCredentialBuilder;
import com.azure.messaging.eventhubs.models.ErrorContext;
import com.azure.messaging.eventhubs.models.EventContext;
import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
@@ -43,7 +44,7 @@ public class EventProcessorTest {
assertThrows(IllegalArgumentException.class,
() -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
- configuration.setBlobAccountName("testAcc");
+ configuration.setBlobAccountName("testAccount");
assertThrows(IllegalArgumentException.class,
() -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
@@ -54,4 +55,30 @@ public class EventProcessorTest {
assertThrows(IllegalArgumentException.class,
() -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
}
+
+ @Test
+ public void testCreateEventProcessorWithTokenCredential() {
+ final EventHubsConfiguration configuration = new EventHubsConfiguration();
+ final Consumer<EventContext> onEvent = event -> {
+ };
+ final Consumer<ErrorContext> onError = error -> {
+ };
+
+ configuration.setBlobContainerName("testContainer");
+ configuration.setBlobAccountName("testAccount");
+ configuration.setBlobAccessKey("testAccess");
+ assertNotNull(EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setTokenCredential(new IntelliJCredentialBuilder().tenantId("tenantId").build());
+ assertThrows(IllegalArgumentException.class,
+ () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setNamespace("namespace");
+ assertThrows(IllegalArgumentException.class,
+ () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setEventHubName("eventHubName");
+ assertNotNull(EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+ }
+
}
diff --git a/parent/pom.xml b/parent/pom.xml
index ec9130705a2..6e523df4eb7 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -84,16 +84,16 @@
<aws-xray-version>2.6.1</aws-xray-version>
<axiom-version>1.2.14</axiom-version>
<azure-jackson2-version>2.13.3</azure-jackson2-version>
- <azure-cosmos-version>4.29.1</azure-cosmos-version>
- <azure-eventhubs-version>5.11.2</azure-eventhubs-version>
- <azure-identity-version>1.5.1</azure-identity-version>
- <azure-keyvault-secrets-version>4.4.2</azure-keyvault-secrets-version>
- <azure-messaging-eventhubs-checkpointstore-blob-version>1.12.1</azure-messaging-eventhubs-checkpointstore-blob-version>
- <azure-messaging-servicebus-version>7.8.0</azure-messaging-servicebus-version>
- <azure-storage-blob-version>12.16.1</azure-storage-blob-version>
- <azure-storage-blob-changefeed-version>12.0.0-beta.17</azure-storage-blob-changefeed-version>
- <azure-storage-datalake-version>12.9.1</azure-storage-datalake-version>
- <azure-storage-queue-version>12.12.2</azure-storage-queue-version>
+ <azure-cosmos-version>4.35.0</azure-cosmos-version>
+ <azure-eventhubs-version>5.13.0</azure-eventhubs-version>
+ <azure-identity-version>1.5.4</azure-identity-version>
+ <azure-keyvault-secrets-version>4.4.6</azure-keyvault-secrets-version>
+ <azure-messaging-eventhubs-checkpointstore-blob-version>1.15.0</azure-messaging-eventhubs-checkpointstore-blob-version>
+ <azure-messaging-servicebus-version>7.10.1</azure-messaging-servicebus-version>
+ <azure-storage-blob-version>12.19.0</azure-storage-blob-version>
+ <azure-storage-blob-changefeed-version>12.0.0-beta.18</azure-storage-blob-changefeed-version>
+ <azure-storage-datalake-version>12.12.0</azure-storage-datalake-version>
+ <azure-storage-queue-version>12.14.1</azure-storage-queue-version>
<bouncycastle-version>1.71</bouncycastle-version>
<box-java-sdk-version>2.58.0</box-java-sdk-version>
<braintree-gateway-version>3.16.1</braintree-gateway-version>
diff --git a/test-infra/camel-test-infra-azure-common/src/test/java/org/apache/camel/test/infra/azure/common/services/AzuriteContainer.java b/test-infra/camel-test-infra-azure-common/src/test/java/org/apache/camel/test/infra/azure/common/services/AzuriteContainer.java
index 09ac0c7d0fc..4d4360fdef9 100644
--- a/test-infra/camel-test-infra-azure-common/src/test/java/org/apache/camel/test/infra/azure/common/services/AzuriteContainer.java
+++ b/test-infra/camel-test-infra-azure-common/src/test/java/org/apache/camel/test/infra/azure/common/services/AzuriteContainer.java
@@ -26,7 +26,7 @@ public class AzuriteContainer extends GenericContainer<AzuriteContainer> {
public static final String DEFAULT_ACCOUNT_KEY
= "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
- public static final String IMAGE_NAME = "mcr.microsoft.com/azure-storage/azurite:3.16.0";
+ public static final String IMAGE_NAME = "mcr.microsoft.com/azure-storage/azurite:3.19.0";
public AzuriteContainer() {
this(IMAGE_NAME);