This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 04cf4bd8a628 CAMEL-25660: camel-azure-servicebus - bound the
dead-letter reason and description, and report a rejected request (#27834)
04cf4bd8a628 is described below
commit 04cf4bd8a6282237fb4bace1bb9143982ce5b1bc
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Oct 15 16:38:32 2026 +0200
CAMEL-25660: camel-azure-servicebus - bound the dead-letter reason and
description, and report a rejected request (#27834)
Co-Authored-By: Claude Opus 5 <[email protected]>
---
.../azure/servicebus/ServiceBusConsumer.java | 67 ++++++++++++---
.../azure/servicebus/ServiceBusConsumerTest.java | 97 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++
3 files changed, 163 insertions(+), 10 deletions(-)
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 2bfd5ec6f374..ca9d65d9f710 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
@@ -39,6 +39,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.spi.ExceptionHandler;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.spi.PeriodTaskScheduler;
import org.apache.camel.spi.ShutdownAware;
@@ -46,6 +47,7 @@ import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.StringHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,6 +55,15 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
private static final Logger LOG =
LoggerFactory.getLogger(ServiceBusConsumer.class);
private static final int LOCK_RENEW_INTERVAL_SECONDS = 10;
+ // Azure recommends putting the exception type in the dead-letter reason
and its stack trace in the description,
+ // but both are length-limited: the .NET and JavaScript SDKs cap them at
4096 (MaxDeadLetterReasonLength, applied
+ // to the description as well), and Azure separately warns that a stack
trace here can push a message past the
+ // 256 KB quota of the Standard tier. The Java SDK does not check, so an
over-long value only fails at the broker -
+ // which leaves the message locked. Cap it here instead: the head of a
stack trace carries the cause, the tail is
+ // error-handler frames.
+ private static final int MAX_DEAD_LETTER_TEXT_LENGTH = 4096;
+ // so that whoever reads the dead-letter queue can see the text was cut
rather than the trace having ended
+ private static final String TRUNCATED_SUFFIX = "...";
private ServiceBusProcessorClient client;
private boolean clientCloseable;
@@ -385,25 +396,61 @@ public class ServiceBusConsumer extends DefaultConsumer
implements ShutdownAware
if (getConfiguration().getServiceBusReceiveMode() ==
ServiceBusReceiveMode.PEEK_LOCK) {
if (getConfiguration().isEnableDeadLettering() &&
(ObjectHelper.isEmpty(getConfiguration().getSubQueue())
||
ObjectHelper.equal(getConfiguration().getSubQueue(), SubQueue.NONE))) {
- DeadLetterOptions deadLetterOptions = new
DeadLetterOptions();
if (cause != null) {
- deadLetterOptions
- .setDeadLetterReason(
- String.format("%s: %s",
cause.getClass().getName(), cause.getMessage()));
-
deadLetterOptions.setDeadLetterErrorDescription(Arrays.stream(cause.getStackTrace())
- .map(StackTraceElement::toString)
- .collect(Collectors.joining("\n")));
- messageContext.deadLetter(deadLetterOptions);
+ DeadLetterOptions options =
deadLetterOptions(cause);
+ settle(exchange, "dead-letter", () ->
messageContext.deadLetter(options));
} else {
- messageContext.deadLetter();
+ settle(exchange, "dead-letter",
messageContext::deadLetter);
}
} else {
- messageContext.abandon();
+ settle(exchange, "abandon", messageContext::abandon);
}
}
} finally {
pendingExchanges.decrementAndGet();
}
}
+
+ /**
+ * Describes the failure for the dead-letter queue, bounded so that
the diagnostics cannot be what makes the
+ * broker reject the message.
+ */
+ private DeadLetterOptions deadLetterOptions(Exception cause) {
+ String stackTrace = Arrays.stream(cause.getStackTrace())
+ .map(StackTraceElement::toString)
+ .collect(Collectors.joining("\n"));
+
+ return new DeadLetterOptions()
+ .setDeadLetterReason(
+ limitToDeadLetterLength(String.format("%s: %s",
cause.getClass().getName(),
+ cause.getMessage())))
+
.setDeadLetterErrorDescription(limitToDeadLetterLength(stackTrace));
+ }
+
+ /**
+ * Caps a dead-letter field, leaving room for the suffix so the result
still fits.
+ */
+ private String limitToDeadLetterLength(String value) {
+ return StringHelper.limitLength(value, MAX_DEAD_LETTER_TEXT_LENGTH
- TRUNCATED_SUFFIX.length(),
+ TRUNCATED_SUFFIX);
+ }
+
+ /**
+ * Settles the message, reporting a rejected request through the
consumer's {@link ExceptionHandler}.
+ * <p/>
+ * Without this the exception only reaches {@code
UnitOfWorkHelper.doneSynchronization}, which logs it as
+ * "Exception occurred during onCompletion. This exception will be
ignored." - so the operator sees a generic
+ * core warning rather than a message from this consumer, and nothing
routable. Either way the message is not
+ * settled: it stays locked until the lock expires, Service Bus
redelivers it, and it ends up dead-lettered with
+ * {@code MaxDeliveryCountExceeded} instead of the reason collected
here.
+ */
+ private void settle(Exchange exchange, String operation, Runnable
settlement) {
+ try {
+ settlement.run();
+ } catch (Exception e) {
+ getExceptionHandler().handleException(
+ "Cannot " + operation + " the message. It stays locked
and will be redelivered.", exchange, e);
+ }
+ }
}
}
diff --git
a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
index d63ee50844e1..d490e0ff2e67 100644
---
a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
+++
b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
@@ -385,6 +385,103 @@ public class ServiceBusConsumerTest {
}
}
+ @Test
+ void synchronizationBoundsTheDeadLetterReasonAndDescription() throws
Exception {
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.isEnableDeadLettering()).thenReturn(true);
+
+ consumer.doStart();
+ when(messageContext.getMessage()).thenReturn(message);
+ processMessageCaptor.getValue().accept(messageContext);
+
+ Exchange exchange = exchangeCaptor.getValue();
+
exchange.setException(exceptionWithALongMessageAndADeepStackTrace());
+
+ Synchronization synchronization =
exchange.getExchangeExtension().handoverCompletions().get(0);
+ synchronization.onFailure(exchange);
+
+ ArgumentCaptor<DeadLetterOptions> captor = ArgumentCaptor.captor();
+ verify(messageContext).deadLetter(captor.capture());
+ DeadLetterOptions options = captor.getValue();
+
+ // the .NET and JavaScript SDKs cap both fields at 4096
(MaxDeadLetterReasonLength, applied to the
+ // description too); the Java SDK does not check, so an over-long
value would only fail at the broker
+ assertThat(options.getDeadLetterReason())
+ .hasSizeLessThanOrEqualTo(4096)
+ .startsWith(IllegalStateException.class.getName())
+ .endsWith("...");
+ assertThat(options.getDeadLetterErrorDescription())
+ .hasSizeLessThanOrEqualTo(4096)
+ .contains(getClass().getName())
+ .endsWith("...");
+ }
+ }
+
+ @Test
+ void synchronizationReportsARejectedDeadLetterInsteadOfThrowing() throws
Exception {
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ when(configuration.isEnableDeadLettering()).thenReturn(true);
+ consumer.setExceptionHandler(exceptionHandler);
+
+ consumer.doStart();
+ when(messageContext.getMessage()).thenReturn(message);
+ processMessageCaptor.getValue().accept(messageContext);
+
+ Exchange exchange = exchangeCaptor.getValue();
+ final Exception routeFailure = new Exception("Test exception");
+ exchange.setException(routeFailure);
+
+ // the broker refuses the request, for instance because the
message is over the size quota
+ final RuntimeException rejected = new
IllegalStateException("Dead-lettering rejected");
+
doThrow(rejected).when(messageContext).deadLetter(any(DeadLetterOptions.class));
+
+ Synchronization synchronization =
exchange.getExchangeExtension().handoverCompletions().get(0);
+ // must not escape into the SDK's processor callback, where
nothing would report it
+ synchronization.onFailure(exchange);
+
+ verify(exceptionHandler).handleException(anyString(),
eq(exchange), eq(rejected));
+ }
+ }
+
+ @Test
+ void synchronizationReportsARejectedAbandonInsteadOfThrowing() throws
Exception {
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
+
when(configuration.getServiceBusReceiveMode()).thenReturn(ServiceBusReceiveMode.PEEK_LOCK);
+ // dead-lettering off, so a failed exchange is abandoned instead
+ when(configuration.isEnableDeadLettering()).thenReturn(false);
+ consumer.setExceptionHandler(exceptionHandler);
+
+ consumer.doStart();
+ when(messageContext.getMessage()).thenReturn(message);
+ processMessageCaptor.getValue().accept(messageContext);
+
+ Exchange exchange = exchangeCaptor.getValue();
+ exchange.setException(new Exception("Test exception"));
+
+ final RuntimeException rejected = new
IllegalStateException("Abandon rejected");
+ doThrow(rejected).when(messageContext).abandon();
+
+ Synchronization synchronization =
exchange.getExchangeExtension().handoverCompletions().get(0);
+ synchronization.onFailure(exchange);
+
+ verify(exceptionHandler).handleException(anyString(),
eq(exchange), eq(rejected));
+ }
+ }
+
+ private static Exception exceptionWithALongMessageAndADeepStackTrace() {
+ Exception answer = new IllegalStateException("x".repeat(8192));
+ StackTraceElement[] frames = new StackTraceElement[512];
+ for (int i = 0; i < frames.length; i++) {
+ frames[i] = new StackTraceElement(
+ ServiceBusConsumerTest.class.getName(), "frame" + i,
+ "ServiceBusConsumerTest.java", i);
+ }
+ answer.setStackTrace(frames);
+ return answer;
+ }
+
@Test
void
synchronizationDeadLettersMessageWithoutOptionsWhenExceptionNotPresent() throws
Exception {
try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint,
processor)) {
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 45283da3e7cd..ebcf624762ff 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -734,6 +734,15 @@ go through one of those paths, remove the `{` and `}`
characters from it in the
endpoint (Chunk has no escape for tags inside data), for example with a
processor that rewrites the body or
header.
+=== camel-azure-servicebus - the dead-letter reason and description are bounded
+
+When the consumer dead-letters a message after a failed route with
`enableDeadLettering=true`, the reason and the
+error description it writes are now capped at 4096 characters, the limit the
Azure SDKs apply to both fields. A value
+that was cut ends with `...`. The description holds the stack trace, so a deep
one is now truncated rather than sent
+in full and rejected by the broker.
+
+A settlement request the broker rejects is now reported through the consumer's
exception handler.
+
=== camel-telegram - a failed send fails the exchange once
When Telegram answers a message with an error status (for example `400` for an
unknown chat), or with a body that
