This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 449b0348ff4e [backport camel-4.22.x] CAMEL-25227:
camel-azure-eventhubs - partition headers and default header filter (#27378)
449b0348ff4e is described below
commit 449b0348ff4efdbf4647338a7f4d7a72bcaa8dea
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Oct 5 19:50:54 2026 +0200
[backport camel-4.22.x] CAMEL-25227: camel-azure-eventhubs - partition
headers and default header filter (#27378)
CAMEL-25227: camel-azure-eventhubs - partition headers and default header
filter (#27213)
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../catalog/docs/azure-eventhubs-component.adoc | 16 ++
.../src/main/docs/azure-eventhubs-component.adoc | 16 ++
.../azure/eventhubs/EventHubsComponent.java | 11 ++
.../EventHubsConfigurationOptionsProxy.java | 31 +++-
.../azure/eventhubs/EventHubsConstants.java | 5 +
.../azure/eventhubs/EventHubsConsumer.java | 19 +-
.../operations/EventHubsProducerOperations.java | 11 +-
.../eventhubs/EventHubsConsumerToProducerTest.java | 201 +++++++++++++++++++++
8 files changed, 296 insertions(+), 14 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 aac3b8fb8d51..79cb97975367 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
@@ -184,6 +184,22 @@ from("direct:start")
.to("azure-eventhubs:?connectionString=RAW({{connectionString}})"
----
+The producer sends the events to the partition chosen, in this order, by:
+
+. a `CamelAzureEventHubsPartitionKey` or `CamelAzureEventHubsPartitionId`
header set by the route
+. the `partitionKey` or `partitionId` endpoint option
+. the partition key of the event received by an `azure-eventhubs` consumer
earlier in the route, so that events with
+the same key stay together
+
+Otherwise, Event Hubs picks the partition. The consumer sets the same two
headers on every event it receives. While
+they still hold the received values, the producer does not take them as a
partition chosen by the route, and it never
+reuses the partition id of a received event, because partitions differ from
one Event Hub to another.
+
+To keep the partition of the received event, for example when mirroring
between Event Hubs that have the same
+partitions, remove the `CamelAzureEventHubsReceivedPartitionId` exchange
property, or
+`CamelAzureEventHubsReceivedPartitionKey` for the partition key, before the
producer, such as with
+`.removeProperty(EventHubsConstants.RECEIVED_PARTITION_ID)`. The header then
counts as chosen by the route.
+
=== 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]
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 aac3b8fb8d51..79cb97975367 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
@@ -184,6 +184,22 @@ from("direct:start")
.to("azure-eventhubs:?connectionString=RAW({{connectionString}})"
----
+The producer sends the events to the partition chosen, in this order, by:
+
+. a `CamelAzureEventHubsPartitionKey` or `CamelAzureEventHubsPartitionId`
header set by the route
+. the `partitionKey` or `partitionId` endpoint option
+. the partition key of the event received by an `azure-eventhubs` consumer
earlier in the route, so that events with
+the same key stay together
+
+Otherwise, Event Hubs picks the partition. The consumer sets the same two
headers on every event it receives. While
+they still hold the received values, the producer does not take them as a
partition chosen by the route, and it never
+reuses the partition id of a received event, because partitions differ from
one Event Hub to another.
+
+To keep the partition of the received event, for example when mirroring
between Event Hubs that have the same
+partitions, remove the `CamelAzureEventHubsReceivedPartitionId` exchange
property, or
+`CamelAzureEventHubsReceivedPartitionKey` for the partition key, before the
producer, such as with
+`.removeProperty(EventHubsConstants.RECEIVED_PARTITION_ID)`. The header then
counts as chosen by the route.
+
=== 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]
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 2a20bccd7f47..4ddd574e5753 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
@@ -23,6 +23,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.component.azure.common.CredentialType;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultHeaderFilterStrategy;
import org.apache.camel.support.HeaderFilterStrategyComponent;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -42,6 +43,16 @@ public class EventHubsComponent extends
HeaderFilterStrategyComponent {
public EventHubsComponent() {
}
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+
+ // keep the Camel-internal headers out of the EventData application
properties
+ if (getHeaderFilterStrategy() == null) {
+ setHeaderFilterStrategy(new DefaultHeaderFilterStrategy());
+ }
+ }
+
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
final EventHubsConfiguration configuration = this.configuration.copy();
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
index 05091b061ccf..84d4c7755bbc 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
@@ -38,11 +38,22 @@ public class EventHubsConfigurationOptionsProxy {
}
public String getPartitionKey(final Exchange exchange) {
- return getOption(exchange, EventHubsConstants.PARTITION_KEY,
configuration::getPartitionKey, String.class);
+ return getOption(exchange, EventHubsConstants.PARTITION_KEY,
EventHubsConstants.RECEIVED_PARTITION_KEY,
+ configuration::getPartitionKey, String.class);
}
public String getPartitionId(final Exchange exchange) {
- return getOption(exchange, EventHubsConstants.PARTITION_ID,
configuration::getPartitionId, String.class);
+ return getOption(exchange, EventHubsConstants.PARTITION_ID,
EventHubsConstants.RECEIVED_PARTITION_ID,
+ configuration::getPartitionId, String.class);
+ }
+
+ /**
+ * The partition key of the event received by an azure-eventhubs consumer,
when the exchange comes from one.
+ */
+ public String getReceivedPartitionKey(final Exchange exchange) {
+ return ObjectHelper.isEmpty(exchange)
+ ? null
+ :
exchange.getProperty(EventHubsConstants.RECEIVED_PARTITION_KEY, String.class);
}
public EventHubsConfiguration getConfiguration() {
@@ -50,11 +61,19 @@ public class EventHubsConfigurationOptionsProxy {
}
private <R> R getOption(
- final Exchange exchange, final String headerName, final
Supplier<R> fallbackFn, final Class<R> type) {
+ final Exchange exchange, final String headerName, final String
receivedPropertyName,
+ final Supplier<R> fallbackFn, final Class<R> type) {
// we first try to look if our value in exchange otherwise fallback to
fallbackFn which could be either a function or constant
- return ObjectHelper.isEmpty(exchange) ||
ObjectHelper.isEmpty(getObjectFromHeaders(exchange, headerName, type))
- ? fallbackFn.get()
- : getObjectFromHeaders(exchange, headerName, type);
+ if (ObjectHelper.isEmpty(exchange)) {
+ return fallbackFn.get();
+ }
+ final R value = getObjectFromHeaders(exchange, headerName, type);
+ // a header still holding the value of the received event describes
that event, it is not a partition
+ // chosen by the route
+ if (ObjectHelper.isEmpty(value) ||
value.equals(exchange.getProperty(receivedPropertyName, type))) {
+ return fallbackFn.get();
+ }
+ return value;
}
}
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
index 3cad12dbc799..b9c230321710 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
@@ -37,6 +37,11 @@ public final class EventHubsConstants {
"(consumer) It sets the partition id of the Event
Hub.",
javaType = "String")
public static final String PARTITION_ID = HEADER_PREFIX + "PartitionId";
+ // exchange properties set by the consumer with the partition of the
received event, so the producer can tell
+ // the headers above apart from a partition chosen by the route. A route
that wants to keep the received
+ // partition, for example when mirroring between Event Hubs with the same
partitions, removes the property.
+ public static final String RECEIVED_PARTITION_KEY = HEADER_PREFIX +
"ReceivedPartitionKey";
+ public static final String RECEIVED_PARTITION_ID = HEADER_PREFIX +
"ReceivedPartitionId";
// headers set by the consumer only
@Metadata(label = "consumer",
description = "It sets the offset of the event when it was
received from the associated Event Hub partition. This is only present on a
received `EventData`.",
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 09345bdb8930..7f0ca1bf3259 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
@@ -132,16 +132,21 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
return (EventHubsEndpoint) super.getEndpoint();
}
- private Exchange createAzureEventHubExchange(final EventContext
eventContext) {
+ Exchange createAzureEventHubExchange(final EventContext eventContext) {
final Exchange exchange = createExchange(true);
final Message message = exchange.getIn();
// set body as byte[] and let camel typeConverters do the job to
convert
message.setBody(eventContext.getEventData().getBody());
- // set headers
- message.setHeader(EventHubsConstants.PARTITION_ID,
eventContext.getPartitionContext().getPartitionId());
- if (eventContext.getEventData().getPartitionKey() != null) {
- message.setHeader(EventHubsConstants.PARTITION_KEY,
eventContext.getEventData().getPartitionKey());
+ // set headers, and record the partition of the received event so an
azure-eventhubs producer later in the
+ // route does not take these headers as a partition chosen by the route
+ final String partitionId =
eventContext.getPartitionContext().getPartitionId();
+ final String partitionKey =
eventContext.getEventData().getPartitionKey();
+ message.setHeader(EventHubsConstants.PARTITION_ID, partitionId);
+ exchange.setProperty(EventHubsConstants.RECEIVED_PARTITION_ID,
partitionId);
+ if (partitionKey != null) {
+ message.setHeader(EventHubsConstants.PARTITION_KEY, partitionKey);
+ exchange.setProperty(EventHubsConstants.RECEIVED_PARTITION_KEY,
partitionKey);
}
message.setHeader(EventHubsConstants.OFFSET,
eventContext.getEventData().getOffset());
message.setHeader(EventHubsConstants.ENQUEUED_TIME,
eventContext.getEventData().getEnqueuedTime());
@@ -160,7 +165,9 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
final Message message = exchange.getIn();
// set headers
- message.setHeader(EventHubsConstants.PARTITION_ID,
errorContext.getPartitionContext().getPartitionId());
+ final String partitionId =
errorContext.getPartitionContext().getPartitionId();
+ message.setHeader(EventHubsConstants.PARTITION_ID, partitionId);
+ exchange.setProperty(EventHubsConstants.RECEIVED_PARTITION_ID,
partitionId);
// set exception
exchange.setException(errorContext.getThrowable());
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
index 8d70509b23b8..fde3cb0dee71 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
@@ -58,8 +58,15 @@ public class EventHubsProducerOperations {
ObjectHelper.notNull(exchange, "exchange cannot be null");
ObjectHelper.notNull(callback, "callback cannot be null");
- final SendOptions sendOptions =
createSendOptions(configurationOptionsProxy.getPartitionKey(exchange),
- configurationOptionsProxy.getPartitionId(exchange));
+ String partitionKey =
configurationOptionsProxy.getPartitionKey(exchange);
+ final String partitionId =
configurationOptionsProxy.getPartitionId(exchange);
+ if (ObjectHelper.isEmpty(partitionKey) &&
ObjectHelper.isEmpty(partitionId)) {
+ // neither the route nor the endpoint chose a partition: keep the
partition key of a received event,
+ // so events with the same key stay together, but never its
partition id, as partitions differ between
+ // Event Hubs
+ partitionKey =
configurationOptionsProxy.getReceivedPartitionKey(exchange);
+ }
+ final SendOptions sendOptions = createSendOptions(partitionKey,
partitionId);
final Iterable<EventData> eventData = createEventData(exchange);
return sendAsyncEvents(eventData, sendOptions, exchange, callback);
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerToProducerTest.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerToProducerTest.java
new file mode 100644
index 000000000000..e1ec424d2074
--- /dev/null
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerToProducerTest.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.component.azure.eventhubs;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+
+import com.azure.messaging.eventhubs.EventData;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.models.EventContext;
+import com.azure.messaging.eventhubs.models.PartitionContext;
+import com.azure.messaging.eventhubs.models.SendOptions;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.ExchangeBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Mono;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyIterable;
+import static org.mockito.Mockito.atMostOnce;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * An exchange created by the azure-eventhubs consumer and sent to an
azure-eventhubs producer, as in a route that
+ * consumes from one Event Hub and produces to another.
+ */
+class EventHubsConsumerToProducerTest extends CamelTestSupport {
+
+ private static final String CONSUMER =
"azure-eventhubs:namespace/hub?connectionString=RAW(Endpoint=sb://"
+ +
"namespace.servicebus.windows.net/;SharedAccessKeyName=name;SharedAccessKey=a2V5;EntityPath=hub)";
+ private static final String PRODUCER =
"azure-eventhubs:?producerAsyncClient=#producerClient";
+
+ private final EventHubProducerAsyncClient producerClient =
mock(EventHubProducerAsyncClient.class);
+
+ @BeforeEach
+ void bindProducerClient() {
+ when(producerClient.send(anyIterable())).thenReturn(Mono.empty());
+ when(producerClient.send(anyIterable(),
any(SendOptions.class))).thenReturn(Mono.empty());
+ context.getRegistry().bind("producerClient", producerClient);
+ }
+
+ @Test
+ void consumerRecordsThePartitionOfTheReceivedEvent() {
+ Exchange exchange = received("3", "device-1");
+
+ assertEquals("3",
exchange.getMessage().getHeader(EventHubsConstants.PARTITION_ID));
+ assertEquals("device-1",
exchange.getMessage().getHeader(EventHubsConstants.PARTITION_KEY));
+ assertEquals("3",
exchange.getProperty(EventHubsConstants.RECEIVED_PARTITION_ID));
+ assertEquals("device-1",
exchange.getProperty(EventHubsConstants.RECEIVED_PARTITION_KEY));
+ }
+
+ @Test
+ void receivedPartitionKeyIsKeptWhenNoPartitionIsChosen() {
+ SendOptions options = send(PRODUCER, received("3", "device-1"));
+
+ assertEquals("device-1", options.getPartitionKey());
+ assertNull(options.getPartitionId());
+ }
+
+ @Test
+ void receivedPartitionIdIsNotReused() {
+ assertNull(send(PRODUCER, received("3", null)));
+ verify(producerClient).send(anyIterable());
+ }
+
+ @Test
+ void configuredPartitionKeyIsUsed() {
+ String producer = PRODUCER + "&partitionKey=configured";
+
+ assertEquals("configured", send(producer, received("3",
"device-1")).getPartitionKey());
+ assertEquals("configured", send(producer, received("3",
null)).getPartitionKey());
+ }
+
+ @Test
+ void configuredPartitionIdIsUsed() {
+ String producer = PRODUCER + "&partitionId=1";
+
+ SendOptions keyed = send(producer, received("3", "device-1"));
+ assertNull(keyed.getPartitionKey());
+ assertEquals("1", keyed.getPartitionId());
+
+ SendOptions unkeyed = send(producer, received("3", null));
+ assertNull(unkeyed.getPartitionKey());
+ assertEquals("1", unkeyed.getPartitionId());
+ }
+
+ @Test
+ void partitionChosenByTheRouteIsUsed() {
+ Exchange rekeyed = received("3", "device-1");
+ rekeyed.getMessage().setHeader(EventHubsConstants.PARTITION_KEY,
"tenant-7");
+ SendOptions byKey = send(PRODUCER, rekeyed);
+ assertEquals("tenant-7", byKey.getPartitionKey());
+ assertNull(byKey.getPartitionId());
+
+ Exchange pinned = received("3", "device-1");
+ pinned.getMessage().setHeader(EventHubsConstants.PARTITION_ID, "1");
+ SendOptions byId = send(PRODUCER, pinned);
+ assertNull(byId.getPartitionKey());
+ assertEquals("1", byId.getPartitionId());
+ }
+
+ @Test
+ void receivedPartitionIsKeptWhenTheRouteRemovesTheReceivedProperty() {
+ Exchange mirrored = received("3", null);
+ mirrored.removeProperty(EventHubsConstants.RECEIVED_PARTITION_ID);
+ SendOptions byId = send(PRODUCER, mirrored);
+ assertNull(byId.getPartitionKey());
+ assertEquals("3", byId.getPartitionId());
+
+ Exchange keyed = received("3", "device-1");
+ keyed.removeProperty(EventHubsConstants.RECEIVED_PARTITION_KEY);
+ SendOptions byKey = send(PRODUCER + "&partitionKey=configured", keyed);
+ assertEquals("device-1", byKey.getPartitionKey());
+ assertNull(byKey.getPartitionId());
+ }
+
+ @Test
+ void headerOverridesTheEndpointOptionWhenNotFromConsumer() {
+ Exchange exchange =
ExchangeBuilder.anExchange(context).withBody("event")
+ .withHeader(EventHubsConstants.PARTITION_ID, "1").build();
+
+ assertEquals("1", send(PRODUCER + "&partitionId=2",
exchange).getPartitionId());
+ }
+
+ @Test
+ void partitionKeyAndIdSetTogetherAreRejectedWhenNotFromConsumer() {
+ Exchange result = template.send(PRODUCER,
ExchangeBuilder.anExchange(context).withBody("event")
+ .withHeader(EventHubsConstants.PARTITION_KEY, "device-1")
+ .withHeader(EventHubsConstants.PARTITION_ID, "1").build());
+
+ assertInstanceOf(IllegalArgumentException.class,
result.getException());
+ }
+
+ @Test
+ void camelHeadersAreNotSentAsEventProperties() {
+ Exchange exchange = received("3", "device-1");
+ exchange.getMessage().setHeader("tenant", "acme");
+
+ send(PRODUCER, exchange);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Iterable<EventData>> events =
ArgumentCaptor.forClass(Iterable.class);
+ verify(producerClient).send(events.capture(), any(SendOptions.class));
+ EventData sent = events.getValue().iterator().next();
+ assertEquals("acme", sent.getProperties().get("tenant"));
+ assertTrue(sent.getProperties().keySet().stream().noneMatch(name ->
name.startsWith("Camel")),
+ "Camel headers sent as event properties: " +
sent.getProperties().keySet());
+ }
+
+ private SendOptions send(String uri, Exchange exchange) {
+ clearInvocations(producerClient);
+ Exchange result = template.send(uri, exchange);
+ assertNull(result.getException());
+
+ ArgumentCaptor<SendOptions> options =
ArgumentCaptor.forClass(SendOptions.class);
+ verify(producerClient, atMostOnce()).send(anyIterable(),
options.capture());
+ return options.getAllValues().isEmpty() ? null : options.getValue();
+ }
+
+ private Exchange received(String partitionId, String partitionKey) {
+ EventData eventData = mock(EventData.class);
+
when(eventData.getBody()).thenReturn("event".getBytes(StandardCharsets.UTF_8));
+ when(eventData.getPartitionKey()).thenReturn(partitionKey);
+ when(eventData.getEnqueuedTime()).thenReturn(Instant.now());
+ PartitionContext partitionContext = mock(PartitionContext.class);
+ when(partitionContext.getPartitionId()).thenReturn(partitionId);
+ EventContext eventContext = mock(EventContext.class);
+ when(eventContext.getEventData()).thenReturn(eventData);
+ when(eventContext.getPartitionContext()).thenReturn(partitionContext);
+
+ EventHubsConsumer consumer = new EventHubsConsumer(
+ context.getEndpoint(CONSUMER, EventHubsEndpoint.class),
+ exchange -> {
+ });
+ return consumer.createAzureEventHubExchange(eventContext);
+ }
+}
