This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24158 in repository https://gitbox.apache.org/repos/asf/camel.git
commit f7449a388fecd13af9a618409e118df6ceba314a Author: Claus Ibsen <[email protected]> AuthorDate: Sat Jul 18 09:50:38 2026 +0200 CAMEL-24158: Fix medium and high-severity bugs across Azure components Blob: fix lease ID lost on discarded BlobRequestConditions, fix non-mark InputStream uploaded as 0-byte blob (add StreamCache/FileInputStream support), fix moveAfterRead copy using beginCopy instead of bare URL copyFromUrl, fix page-blob range off-by-one. ServiceBus: fix processMessage exception leaking pendingExchanges counter and unsettled messages. EventHubs: fix consumer setting both PARTITION_ID and PARTITION_KEY causing producer IllegalArgumentException, add HeaderFilterStrategy to prevent all Camel headers leaking to AMQP application properties. Queue: fix popReceipt clobbering by removing stale header save/restore in completion handler. DataLake: fix openInputStream using Blob Query API instead of standard read, deprecate unused SkipLastByteInputStream. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../eventhubs/EventHubsComponentConfigurer.java | 6 ++++ .../component/azure/eventhubs/azure-eventhubs.json | 11 ++++--- .../azure/eventhubs/EventHubsComponent.java | 4 +-- .../azure/eventhubs/EventHubsConsumer.java | 4 ++- .../azure/eventhubs/EventHubsProducer.java | 5 ++- .../operations/EventHubsProducerOperations.java | 36 +++++++++++++++------- .../operations/EventHubsProducerOperationsIT.java | 9 ++++-- .../azure/servicebus/ServiceBusConsumer.java | 34 +++++++++++++------- .../component/azure/storage/blob/BlobBlock.java | 7 ++++- .../blob/BlobConfigurationOptionsProxy.java | 2 +- .../component/azure/storage/blob/BlobConsumer.java | 10 ++++-- .../azure/storage/blob/BlobStreamAndLength.java | 13 ++++++-- .../component/azure/storage/blob/BlobUtils.java | 10 +++--- .../storage/blob/operations/BlobOperations.java | 8 ++--- .../datalake/client/DataLakeFileClientWrapper.java | 6 +--- .../azure/storage/queue/QueueConsumer.java | 12 -------- .../apache/camel/util/SkipLastByteInputStream.java | 3 ++ 17 files changed, 114 insertions(+), 66 deletions(-) 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 4cf39976fd6c..809c16d4fac4 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 @@ -61,6 +61,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl case "credentialType": getOrCreateConfiguration(target).setCredentialType(property(camelContext, org.apache.camel.component.azure.eventhubs.CredentialType.class, value)); return true; case "eventposition": case "eventPosition": getOrCreateConfiguration(target).setEventPosition(property(camelContext, java.util.Map.class, value)); return true; + case "headerfilterstrategy": + case "headerFilterStrategy": target.setHeaderFilterStrategy(property(camelContext, org.apache.camel.spi.HeaderFilterStrategy.class, value)); return true; case "lazystartproducer": case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true; case "partitionid": @@ -120,6 +122,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl case "credentialType": return org.apache.camel.component.azure.eventhubs.CredentialType.class; case "eventposition": case "eventPosition": return java.util.Map.class; + case "headerfilterstrategy": + case "headerFilterStrategy": return org.apache.camel.spi.HeaderFilterStrategy.class; case "lazystartproducer": case "lazyStartProducer": return boolean.class; case "partitionid": @@ -175,6 +179,8 @@ public class EventHubsComponentConfigurer extends PropertyConfigurerSupport impl case "credentialType": return getOrCreateConfiguration(target).getCredentialType(); case "eventposition": case "eventPosition": return getOrCreateConfiguration(target).getEventPosition(); + case "headerfilterstrategy": + case "headerFilterStrategy": return target.getHeaderFilterStrategy(); case "lazystartproducer": case "lazyStartProducer": return target.isLazyStartProducer(); case "partitionid": diff --git a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json index 4c5446db52ce..1adb790ebf23 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json +++ b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json @@ -43,11 +43,12 @@ "partitionKey": { "index": 16, "kind": "property", "displayName": "Partition Key", "group": "producer", "label": "producer", "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": "Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs [...] "producerAsyncClient": { "index": 17, "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 Eve [...] "autowiredEnabled": { "index": 18, "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 [...] - "connectionString": { "index": 19, "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccess [...] - "credentialType": { "index": 20, "kind": "property", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "enum", "javaType": "org.apache.camel.component.azure.eventhubs.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfigurat [...] - "sharedAccessKey": { "index": 21, "kind": "property", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName." }, - "sharedAccessName": { "index": 22, "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": { "index": 23, "kind": "property", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Provide custom a [...] + "headerFilterStrategy": { "index": 19, "kind": "property", "displayName": "Header Filter Strategy", "group": "filter", "label": "filter", "required": false, "type": "object", "javaType": "org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message." }, + "connectionString": { "index": 20, "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccess [...] + "credentialType": { "index": 21, "kind": "property", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "enum", "javaType": "org.apache.camel.component.azure.eventhubs.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfigurat [...] + "sharedAccessKey": { "index": 22, "kind": "property", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName." }, + "sharedAccessName": { "index": 23, "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": { "index": 24, "kind": "property", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Provide custom a [...] }, "headers": { "CamelAzureEventHubsPartitionKey": { "index": 0, "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 [...] 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 489894ca9037..8a5040a060b8 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 @@ -22,7 +22,7 @@ import com.azure.identity.DefaultAzureCredential; import org.apache.camel.Endpoint; import org.apache.camel.spi.Metadata; import org.apache.camel.spi.annotations.Component; -import org.apache.camel.support.DefaultComponent; +import org.apache.camel.support.HeaderFilterStrategyComponent; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,7 +31,7 @@ import org.slf4j.LoggerFactory; * Azure EventHubs component */ @Component("azure-eventhubs") -public class EventHubsComponent extends DefaultComponent { +public class EventHubsComponent extends HeaderFilterStrategyComponent { private static final Logger LOG = LoggerFactory.getLogger(EventHubsComponent.class); 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 b066caba6377..2f3c6c3c5827 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 @@ -134,7 +134,9 @@ public class EventHubsConsumer extends DefaultConsumer implements ShutdownAware message.setBody(eventContext.getEventData().getBody()); // set headers message.setHeader(EventHubsConstants.PARTITION_ID, eventContext.getPartitionContext().getPartitionId()); - message.setHeader(EventHubsConstants.PARTITION_KEY, eventContext.getEventData().getPartitionKey()); + if (eventContext.getEventData().getPartitionKey() != null) { + message.setHeader(EventHubsConstants.PARTITION_KEY, eventContext.getEventData().getPartitionKey()); + } message.setHeader(EventHubsConstants.OFFSET, eventContext.getEventData().getOffset()); message.setHeader(EventHubsConstants.ENQUEUED_TIME, eventContext.getEventData().getEnqueuedTime()); message.setHeader(EventHubsConstants.SEQUENCE_NUMBER, eventContext.getEventData().getSequenceNumber()); 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 2c5913a0bf75..61d345dd88ba 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 @@ -22,6 +22,7 @@ import org.apache.camel.Endpoint; import org.apache.camel.Exchange; import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory; import org.apache.camel.component.azure.eventhubs.operations.EventHubsProducerOperations; +import org.apache.camel.spi.HeaderFilterStrategy; import org.apache.camel.support.DefaultAsyncProducer; public class EventHubsProducer extends DefaultAsyncProducer { @@ -48,7 +49,9 @@ public class EventHubsProducer extends DefaultAsyncProducer { } // create our operations - producerOperations = new EventHubsProducerOperations(producerAsyncClient, getConfiguration()); + HeaderFilterStrategy headerFilterStrategy + = ((EventHubsComponent) getEndpoint().getComponent()).getHeaderFilterStrategy(); + producerOperations = new EventHubsProducerOperations(producerAsyncClient, getConfiguration(), headerFilterStrategy); } @Override 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 23498d29d0e0..8d70509b23b8 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 @@ -30,6 +30,7 @@ import org.apache.camel.Message; import org.apache.camel.TypeConverter; import org.apache.camel.component.azure.eventhubs.EventHubsConfiguration; import org.apache.camel.component.azure.eventhubs.EventHubsConfigurationOptionsProxy; +import org.apache.camel.spi.HeaderFilterStrategy; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,12 +42,15 @@ public class EventHubsProducerOperations { private final EventHubProducerAsyncClient producerAsyncClient; private final EventHubsConfigurationOptionsProxy configurationOptionsProxy; + private final HeaderFilterStrategy headerFilterStrategy; public EventHubsProducerOperations(final EventHubProducerAsyncClient producerAsyncClient, - final EventHubsConfiguration configuration) { + final EventHubsConfiguration configuration, + final HeaderFilterStrategy headerFilterStrategy) { ObjectHelper.notNull(producerAsyncClient, "client cannot be null"); this.producerAsyncClient = producerAsyncClient; + this.headerFilterStrategy = headerFilterStrategy; configurationOptionsProxy = new EventHubsConfigurationOptionsProxy(configuration); } @@ -107,7 +111,7 @@ public class EventHubsProducerOperations { // check if our exchange is list or contain some values if (exchange.getIn().getBody() instanceof Iterable) { return createEventDataFromIterable((Iterable<Object>) exchange.getIn().getBody(), - exchange.getContext().getTypeConverter(), exchange.getIn().getHeaders()); + exchange.getContext().getTypeConverter(), exchange.getIn().getHeaders(), exchange); } // we have only a single event here @@ -115,16 +119,17 @@ public class EventHubsProducerOperations { } private Iterable<EventData> createEventDataFromIterable( - final Iterable<Object> inputData, final TypeConverter converter, Map<String, Object> headers) { + final Iterable<Object> inputData, final TypeConverter converter, + Map<String, Object> headers, final Exchange exchange) { final List<EventData> finalEventData = new LinkedList<>(); inputData.forEach(data -> { - if (data instanceof Exchange) { - finalEventData.add(createEventDataFromExchange((Exchange) data)); - } else if (data instanceof Message) { - finalEventData.add(createEventDataFromMessage((Message) data)); + if (data instanceof Exchange ex) { + finalEventData.add(createEventDataFromExchange(ex)); + } else if (data instanceof Message msg) { + finalEventData.add(createEventDataFromMessage(msg)); } else { - finalEventData.add(createEventDataFromObject(data, converter, headers)); + finalEventData.add(createEventDataFromObject(data, converter, headers, exchange)); } }); @@ -137,11 +142,12 @@ public class EventHubsProducerOperations { private EventData createEventDataFromMessage(final Message message) { return createEventDataFromObject(message.getBody(), message.getExchange().getContext().getTypeConverter(), - message.getHeaders()); + message.getHeaders(), message.getExchange()); } private EventData createEventDataFromObject( - final Object inputData, final TypeConverter converter, Map<String, Object> headers) { + final Object inputData, final TypeConverter converter, + Map<String, Object> headers, final Exchange exchange) { final byte[] data = converter.convertTo(byte[].class, inputData); if (ObjectHelper.isEmpty(data)) { @@ -152,7 +158,15 @@ public class EventHubsProducerOperations { } EventData eventData = new EventData(data); - eventData.getProperties().putAll(headers); + if (headerFilterStrategy != null) { + headers.forEach((key, value) -> { + if (!headerFilterStrategy.applyFilterToCamelHeaders(key, value, exchange)) { + eventData.getProperties().put(key, value); + } + }); + } else { + eventData.getProperties().putAll(headers); + } return eventData; } diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java index e57ced1a38ad..80786f835149 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java +++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/integration/operations/EventHubsProducerOperationsIT.java @@ -64,7 +64,8 @@ class EventHubsProducerOperationsIT extends CamelTestSupport { @Test public void testSendEventWithSpecificPartition() { - final EventHubsProducerOperations operations = new EventHubsProducerOperations(producerAsyncClient, configuration); + final EventHubsProducerOperations operations + = new EventHubsProducerOperations(producerAsyncClient, configuration, null); final String firstPartition = producerAsyncClient.getPartitionIds().blockLast(); final Exchange exchange = new DefaultExchange(context); @@ -96,7 +97,8 @@ class EventHubsProducerOperationsIT extends CamelTestSupport { @Test public void testIterableExchangesSendEventsWithSpecificPartition() { - final EventHubsProducerOperations operations = new EventHubsProducerOperations(producerAsyncClient, configuration); + final EventHubsProducerOperations operations + = new EventHubsProducerOperations(producerAsyncClient, configuration, null); final String firstPartition = producerAsyncClient.getPartitionIds().blockLast(); final Exchange exchange1 = new DefaultExchange(context); @@ -144,7 +146,8 @@ class EventHubsProducerOperationsIT extends CamelTestSupport { @Test public void testIterableStringSendEventsWithSpecificPartition() { - final EventHubsProducerOperations operations = new EventHubsProducerOperations(producerAsyncClient, configuration); + final EventHubsProducerOperations operations + = new EventHubsProducerOperations(producerAsyncClient, configuration, null); final String firstPartition = producerAsyncClient.getPartitionIds().blockLast(); final List<String> messages = new LinkedList<>(); diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java index 97677316e5b2..2bfd5ec6f374 100644 --- a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java +++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java @@ -119,18 +119,30 @@ public class ServiceBusConsumer extends DefaultConsumer implements ShutdownAware private void processMessage(ServiceBusReceivedMessageContext messageContext) { pendingExchanges.incrementAndGet(); - final ServiceBusReceivedMessage message = messageContext.getMessage(); - final Exchange exchange = createServiceBusExchange(message); - final ConsumerOnCompletion onCompletion = new ConsumerOnCompletion(messageContext); - // add exchange callback - exchange.getExchangeExtension().addOnCompletion(onCompletion); - // track for Camel-managed lock renewal - if (lockRenewer != null) { - lockRenewer.add(exchange, message); + try { + final ServiceBusReceivedMessage message = messageContext.getMessage(); + final Exchange exchange = createServiceBusExchange(message); + final ConsumerOnCompletion onCompletion = new ConsumerOnCompletion(messageContext); + // add exchange callback + exchange.getExchangeExtension().addOnCompletion(onCompletion); + // track for Camel-managed lock renewal + if (lockRenewer != null) { + lockRenewer.add(exchange, message); + } + // use default consumer callback + AsyncCallback cb = defaultConsumerCallback(exchange, true); + getAsyncProcessor().process(exchange, cb); + } catch (Exception e) { + pendingExchanges.decrementAndGet(); + if (getConfiguration().getServiceBusReceiveMode() == ServiceBusReceiveMode.PEEK_LOCK) { + try { + messageContext.abandon(); + } catch (Exception abandonEx) { + // ignore + } + } + getExceptionHandler().handleException("Error processing Service Bus message", e); } - // use default consumer callback - AsyncCallback cb = defaultConsumerCallback(exchange, true); - getAsyncProcessor().process(exchange, cb); } private void processError(ServiceBusErrorContext errorContext) { diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobBlock.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobBlock.java index e079a51614b4..c89a397c2be9 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobBlock.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobBlock.java @@ -16,6 +16,7 @@ */ package org.apache.camel.component.azure.storage.blob; +import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; import java.util.UUID; @@ -37,7 +38,11 @@ public final class BlobBlock { } public static BlobBlock createBlobBlock(final String blockId, final InputStream inputStream) throws IOException { - return createBlobBlock(blockId, BlobUtils.getInputStreamLength(inputStream), inputStream); + InputStream is = inputStream; + if (!is.markSupported()) { + is = new BufferedInputStream(is); + } + return createBlobBlock(blockId, BlobUtils.getInputStreamLength(is), is); } public static BlobBlock createBlobBlock(final String blockId, final long size, final InputStream inputStream) { diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java index e65b3c644e90..bf18e7bcbc58 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java @@ -124,7 +124,7 @@ public class BlobConfigurationOptionsProxy { final PageRange pageRange = getPageRange(exchange); if (pageRange != null) { final long blobOffset = pageRange.getStart(); - final long dataCount = pageRange.getEnd() - pageRange.getStart(); + final long dataCount = pageRange.getEnd() - pageRange.getStart() + 1; return new BlobRange(blobOffset, dataCount); } diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java index ee89c698b0d2..89a16bf70561 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java @@ -17,6 +17,7 @@ package org.apache.camel.component.azure.storage.blob; import java.io.IOException; +import java.time.Duration; import java.util.LinkedList; import java.util.List; import java.util.Queue; @@ -283,8 +284,13 @@ public class BlobConsumer extends ScheduledBatchPollingConsumer { = getEndpoint().getBlobServiceClient().getBlobContainerClient(destinationContainer); BlobClient destBlobClient = destContainerClient.getBlobClient(destinationBlobName); - // Copy from source URL (works within same storage account) - destBlobClient.copyFromUrl(sourceBlobClient.getBlobUrl()); + // Use beginCopy which supports same-account authorization via shared key/Azure AD + Duration timeout = getEndpoint().getConfiguration().getTimeout(); + if (timeout == null) { + timeout = Duration.ofMinutes(5); + } + destBlobClient.beginCopy(sourceBlobClient.getBlobUrl(), Duration.ofSeconds(2)) + .waitForCompletion(timeout); } /** diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobStreamAndLength.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobStreamAndLength.java index d2908a453a04..f650431da603 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobStreamAndLength.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobStreamAndLength.java @@ -52,8 +52,11 @@ public final class BlobStreamAndLength { } if (body instanceof InputStream) { - return new BlobStreamAndLength( - (InputStream) body, blobSize != null ? blobSize : BlobUtils.getInputStreamLength((InputStream) body)); + InputStream is = (InputStream) body; + if (blobSize == null && !is.markSupported()) { + is = new BufferedInputStream(is); + } + return new BlobStreamAndLength(is, blobSize != null ? blobSize : BlobUtils.getInputStreamLength(is)); } if (body instanceof File) { return new BlobStreamAndLength(new BufferedInputStream(new FileInputStream((File) body)), ((File) body).length()); @@ -71,7 +74,11 @@ public final class BlobStreamAndLength { throw new IllegalArgumentException("Unsupported blob type:" + body.getClass().getName()); } - return new BlobStreamAndLength(inputStream, blobSize != null ? blobSize : BlobUtils.getInputStreamLength(inputStream)); + InputStream is = inputStream; + if (blobSize == null && !is.markSupported()) { + is = new BufferedInputStream(is); + } + return new BlobStreamAndLength(is, blobSize != null ? blobSize : BlobUtils.getInputStreamLength(is)); } public InputStream getInputStream() { diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java index ca869ef4434f..7d104fd9f401 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java @@ -17,6 +17,7 @@ package org.apache.camel.component.azure.storage.blob; import java.io.ByteArrayInputStream; +import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; @@ -35,16 +36,17 @@ public final class BlobUtils { } public static long getInputStreamLength(InputStream is) throws IOException { - if (is instanceof StreamCache) { - long len = ((StreamCache) is).length(); + if (is instanceof StreamCache streamCache) { + long len = streamCache.length(); if (len > 0) { return len; } + } else if (is instanceof FileInputStream fis) { + return fis.getChannel().size(); } if (!is.markSupported()) { - // azure cannot use -1 - return 0; + return -1; } if (is instanceof ByteArrayInputStream) { return is.available(); diff --git a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java index 6868f10645c5..6c13d74215aa 100644 --- a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java +++ b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java @@ -811,7 +811,10 @@ public class BlobOperations { final BlobHttpHeaders blobHttpHeaders = configurationProxy.getBlobHttpHeaders(exchange); final Map<String, String> metadata = configurationProxy.getMetadata(exchange); final AccessTier accessTier = configurationProxy.getAccessTier(exchange); - final BlobRequestConditions blobRequestConditions = configurationProxy.getBlobRequestConditions(exchange); + BlobRequestConditions blobRequestConditions = configurationProxy.getBlobRequestConditions(exchange); + if (blobRequestConditions == null) { + blobRequestConditions = new BlobRequestConditions(); + } final Duration timeout = configurationProxy.getTimeout(exchange); final byte[] contentMD5 = configurationProxy.getContentMd5(exchange); @@ -839,9 +842,6 @@ public class BlobOperations { } private BlobLeaseClient acquireLeaseIfConfigured(BlobRequestConditions requestConditions, Exchange exchange) { - if (requestConditions == null) { - requestConditions = new BlobRequestConditions(); - } if (requestConditions.getLeaseId() == null && configurationProxy.getLeaseBlob(exchange)) { BlobLeaseClient leaseClient = client.getLeaseClient(); Integer leaseDurationInSeconds = configurationProxy.getLeaseDurationInSeconds(exchange); diff --git a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java index c7e67aa8a110..d5f888597a39 100644 --- a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java +++ b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java @@ -38,7 +38,6 @@ import com.azure.storage.file.datalake.options.DataLakeFileAppendOptions; import com.azure.storage.file.datalake.options.FileParallelUploadOptions; import com.azure.storage.file.datalake.options.FileQueryOptions; import com.azure.storage.file.datalake.sas.DataLakeServiceSasSignatureValues; -import org.apache.camel.util.SkipLastByteInputStream; public class DataLakeFileClientWrapper { private final DataLakeFileClient client; @@ -64,10 +63,7 @@ public class DataLakeFileClientWrapper { } public InputStream openInputStream() { - String query = "SELECT * from BlobStorage"; - final InputStream sourceInputStream = client.openQueryInputStream(query); - /* Workaround for https://github.com/Azure/azure-sdk-for-java/issues/19612 */ - return new SkipLastByteInputStream(sourceInputStream, (byte) '\n'); + return client.openInputStream().getInputStream(); } public Response<InputStream> openQueryInputStreamWithResponse(final FileQueryOptions queryOptions) { diff --git a/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueConsumer.java b/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueConsumer.java index c5f409890a13..eda1377e0dde 100644 --- a/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueConsumer.java +++ b/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueConsumer.java @@ -17,7 +17,6 @@ package org.apache.camel.component.azure.storage.queue; import java.io.InputStream; -import java.time.Duration; import java.util.LinkedList; import java.util.List; import java.util.Queue; @@ -126,21 +125,10 @@ public class QueueConsumer extends ScheduledBatchPollingConsumer { // update pending number of exchanges pendingExchanges = total - index - 1; - // copy messageId, popReceipt, timeout for fix exchange override case - // azure storage blob can override this headers - final String messageId = exchange.getIn().getHeader(QueueConstants.MESSAGE_ID, String.class); - final String popReceipt = exchange.getIn().getHeader(QueueConstants.POP_RECEIPT, String.class); - final Duration timeout = exchange.getIn().getHeader(QueueConstants.TIMEOUT, Duration.class); - // add on completion to handle after work when the exchange is done exchange.getExchangeExtension().addOnCompletion(new Synchronization() { @Override public void onComplete(Exchange exchange) { - // past messageId, popReceipt, timeout for fix exchange override case - exchange.getIn().setHeader(QueueConstants.MESSAGE_ID, messageId); - exchange.getIn().setHeader(QueueConstants.POP_RECEIPT, popReceipt); - exchange.getIn().setHeader(QueueConstants.TIMEOUT, timeout); - processCommit(exchange); } diff --git a/core/camel-util/src/main/java/org/apache/camel/util/SkipLastByteInputStream.java b/core/camel-util/src/main/java/org/apache/camel/util/SkipLastByteInputStream.java index 53982aca0ad8..c38d0b938372 100644 --- a/core/camel-util/src/main/java/org/apache/camel/util/SkipLastByteInputStream.java +++ b/core/camel-util/src/main/java/org/apache/camel/util/SkipLastByteInputStream.java @@ -23,7 +23,10 @@ import java.io.InputStream; /** * An {@link InputStream} that skips the last byte of the underlying delegate {@link InputStream} if the last byte is * equal to the given {@code matchLast} value. + * + * @deprecated No longer used. Will be removed in a future version. */ +@Deprecated public class SkipLastByteInputStream extends BufferedInputStream { private final byte matchLast;
