diff --git a/components/engine/engine-intent/src/main/resources/java-assistant-guide.md b/components/engine/engine-intent/src/main/resources/java-assistant-guide.md index 5c6250c26b3..19a547bfcd3 100644 --- a/components/engine/engine-intent/src/main/resources/java-assistant-guide.md +++ b/components/engine/engine-intent/src/main/resources/java-assistant-guide.md @@ -119,13 +119,15 @@ completion back. Do not swallow exceptions; do not catch what you cannot handle. - In a **`@Controller`** it becomes an HTTP status the caller sees. - In a **`JobHandler`** it is recorded as a FAILED job-log row and surfaces in the Jobs perspective and the Monitoring shell - a real operational record. -- In a **`MessageHandler`** it is logged with its stack trace and nothing more. The message is - acknowledged and discarded, so there is **no retry and no dead letter**: the work is simply lost. - -So in a listener, treat a throw as a report, never as a recovery. If the work must not be lost, make -it replayable - key it on something durable so a later run can redo it - rather than assuming the -platform will deliver the message again. And always log before you throw there, so the failure has a -message of your own wording and not just a stack trace. +- In a **`MessageHandler`** it is logged with its stack trace and then rethrown to the broker, so the + delivery is **retried** - three further attempts with a backoff - and dead-lettered if they all + fail. There is no job-log row and nothing to re-trigger by hand: the log and the dead-letter queue + are the whole record. + +So in a listener a throw buys you a retry, not an escalation - and a retry only helps if the handler +is safe to run twice. Make the work replayable: key it on something durable so a redelivery after a +partial write completes it instead of duplicating it. And always log before you throw, so the failure +carries a message of your own wording and not just a stack trace. ## The shapes you will actually be asked for diff --git a/components/engine/engine-java/CLAUDE.md b/components/engine/engine-java/CLAUDE.md index 4aa0c5cf64c..cf5666a0158 100644 --- a/components/engine/engine-java/CLAUDE.md +++ b/components/engine/engine-java/CLAUDE.md @@ -143,22 +143,27 @@ mixes them. There is **no** reflective by-name fallback. not.** A `JobHandler` that throws is caught by `JobExecutionService`, recorded as a **FAILED job-log row** and rethrown to Quartz, so the failure is a first-class operational record: it shows up in the Jobs perspective and in the Monitoring shell's failed-jobs tile, and the run can be triggered again. -A `MessageHandler` that throws is **logged with its stack trace** by `ListenerClassConsumer.dispatch` -— and that is all: the JMS session is `AUTO_ACKNOWLEDGE` and the exception never reaches the broker, -so **the message is acknowledged and gone**. No retry, no dead letter, no operational record, nothing -to re-run. (Before that log line existed, a throwing listener produced no output at all — the -handler's own `onError` defaults to a no-op — which made every failure inside generated intent glue -invisible.) +A `MessageHandler` that throws is **logged with its stack trace** by `ListenerClassConsumer.dispatch`, +which then **rethrows so the failure reaches the broker**: the delivery is not acknowledged, and the +bounded redelivery policy the subscription configures (1s initial, 5s, exponential, 3 attempts — the +same budget the JavaScript listener path uses) retries it before the broker dead-letters it. So the +work is retried, but there is still **no job-log row and nothing to trigger by hand** — the log and +the dead-letter queue are the whole operational record. (Before that log line existed, a throwing +listener produced no output at all — the handler's own `onError` defaults to a no-op — and before the +rethrow, the message was acknowledged and the event lost for good.) Two consequences worth internalizing before writing either kind of handler: -- **Do not read a listener throw as recoverable.** The generated templates use the same +- **A listener throw is retried, not escalated.** The generated templates use the same `throw new RuntimeException(…)` idiom in `Job.java.template` and in - `Notification`/`Integration.java.template`; in the job it escalates, in the listener it only - narrates. A developer copying the job pattern into a listener loses the work, not just the alert. -- **Work that must not be lost needs its own arrangement** — an idempotent re-run path keyed on - something durable, or a reconciliation job that finds records left in a pre-handler state. This is - why an event-sourced write in generated glue is written to be replayable rather than transactional. + `Notification`/`Integration.java.template`; in the job it becomes a re-runnable failed row, in the + listener it becomes up to three more attempts and then a dead letter nobody is paged about. Neither + one is a substitute for noticing. +- **A handler must therefore be safe to run twice.** Redelivery means the same message can arrive + again after a partial write, so an event-sourced write in generated glue is written to be + replayable — keyed on something durable, like the posting glue's back-reference — rather than + transactional. Work that must not be lost still wants a reconciliation job that finds records left + in a pre-handler state, because the dead-letter queue is where a poisonous message stops. ## `JavaHandler` (low-level REST) diff --git a/components/engine/engine-java/src/main/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumer.java b/components/engine/engine-java/src/main/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumer.java index 746bde1b9cc..5c61c6cee5d 100644 --- a/components/engine/engine-java/src/main/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumer.java +++ b/components/engine/engine-java/src/main/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumer.java @@ -40,11 +40,13 @@ import org.springframework.stereotype.Component; import jakarta.jms.Connection; +import jakarta.jms.Destination; import jakarta.jms.JMSException; import jakarta.jms.Message; import jakarta.jms.MessageConsumer; import jakarta.jms.Session; import jakarta.jms.TextMessage; +import jakarta.jms.Topic; /** * {@link JavaClassConsumer} that connects client listeners to ActiveMQ queues or topics. Two @@ -274,13 +276,18 @@ private Connection subscribe(Subscription subscription, String scope) { Connection connection = connectionFactory.createConnection( ex -> LOGGER.error("[java-listener] JMS error for [{}]: {}", label, ex.getMessage(), ex), subscriptionId); Session session = connectionFactory.createSession(connection); - // Durable, because this subscription goes down on every republish: the handler set is torn - // down and re-registered, and a plain subscriber would silently lose every event published - // in that window - a record created mid-republish would never start its process, resolve - // its register or post its document, and would look exactly like one the automation had - // nothing to do for. - MessageConsumer consumer = topic ? session.createDurableSubscriber(session.createTopic(destinationName), subscriptionId) - : session.createConsumer(session.createQueue(destinationName)); + Destination destination = topic ? session.createTopic(destinationName) : session.createQueue(destinationName); + // Bound the retries, exactly as the JavaScript listener path does. Without this the broker + // still retries a failed delivery, but on its own defaults rather than a budget this + // project chose - and the two listener paths would disagree about how forgiving they are. + connectionFactory.configureRedeliveryPolicy(connection, destination); + // A topic subscription is also DURABLE, because it goes down on every republish: the handler + // set is torn down and re-registered, and a plain subscriber would silently lose every event + // published in that window - a record created mid-republish would never start its process, + // resolve its register or post its document, and would look exactly like one the automation + // had nothing to do for. The cast is what `topic` already decided one line above. + MessageConsumer consumer = + topic ? session.createDurableSubscriber((Topic) destination, subscriptionId) : session.createConsumer(destination); consumer.setMessageListener(msg -> dispatch(msg, subscription.dispatcher(), label)); LOGGER.info("Java @Listener [{}] connected to {} '{}' for {}.", label, subscription.kind(), destinationName, scope); return connection; @@ -386,12 +393,18 @@ private void dispatch(Message msg, Dispatcher dispatcher, String label) { }); } catch (Exception e) { Throwable cause = e.getCause() != null ? e.getCause() : e; - // This log IS the failure report. The handler's own onError defaults to a no-op, and the - // session is AUTO_ACKNOWLEDGE, so the message is acknowledged and gone either way - without - // this line a throwing handler leaves no trace anywhere: no log, no retry, no dead letter. - // Pass the throwable, not just its message, or the stack trace dies here. + // This log IS the failure report: the handler's own onError defaults to a no-op, so without + // this line a throwing handler leaves no explanation anywhere. Redelivery (below) re-runs + // the work; it does not say why the work failed. Pass the throwable, not just its message, + // or the stack trace dies here. LOGGER.error("@Listener [{}] failed handling a message: {}", label, cause.getMessage(), cause); dispatcher.onError(cause.getMessage(), label); + // Let the failure reach the broker. Swallowing it here acknowledged the message and lost + // the event permanently: no retry, no dead letter, and handlers whose correctness depends + // on a second delivery - the generated posting glue repairs a half-written post on + // redelivery - could never run their repair. Throwing hands the message back so the + // redelivery policy configured above retries it, and dead-letters it once exhausted. + throw new IllegalStateException("@Listener [" + label + "] failed handling a message", cause); } } diff --git a/components/engine/engine-java/src/test/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumerTest.java b/components/engine/engine-java/src/test/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumerTest.java index 9a00f6f53dd..f8921997b18 100644 --- a/components/engine/engine-java/src/test/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumerTest.java +++ b/components/engine/engine-java/src/test/java/org/eclipse/dirigible/engine/java/listener/ListenerClassConsumerTest.java @@ -11,12 +11,14 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -64,6 +66,7 @@ static class RecordingHandler implements MessageHandler { Map observedDuringDispatch; RuntimeException failWith; + final List reportedErrors = new ArrayList<>(); @Override public String destination() { @@ -77,11 +80,19 @@ public void onMessage(String message) { throw failWith; } } + + @Override + public void onError(String error) { + reportedErrors.add(error); + } } private TenantConfigurationService tenantConfigurationService; private RecordingHandler handler; private MessageListener capturedListener; + private ActiveMQConnectionArtifactsFactory connectionFactory; + private Connection connection; + private Queue queue; private ListAppender appender; private Logger consumerLogger; @@ -89,7 +100,7 @@ public void onMessage(String message) { @SuppressWarnings("rawtypes") void setUp() throws Exception { ComponentContainer componentContainer = mock(ComponentContainer.class); - ActiveMQConnectionArtifactsFactory connectionFactory = mock(ActiveMQConnectionArtifactsFactory.class); + connectionFactory = mock(ActiveMQConnectionArtifactsFactory.class); TenantContext tenantContext = mock(TenantContext.class); TenantPropertyManager tenantPropertyManager = mock(TenantPropertyManager.class); tenantConfigurationService = mock(TenantConfigurationService.class); @@ -98,9 +109,9 @@ void setUp() throws Exception { when(componentContainer.instanceOf(RecordingHandler.class)).thenReturn(Optional.of(handler)); - Connection connection = mock(Connection.class); + connection = mock(Connection.class); Session session = mock(Session.class); - Queue queue = mock(Queue.class); + queue = mock(Queue.class); MessageConsumer messageConsumer = mock(MessageConsumer.class); when(connectionFactory.createConnection(any(), any())).thenReturn(connection); when(connectionFactory.createSession(connection)).thenReturn(session); @@ -177,10 +188,11 @@ void theInjectedConfigIsClearedAfterDispatchSoItNeverLeaksOntoThePooledThread() } /** - * A handler that throws must leave a trace. The message is acknowledged and discarded either way - * (AUTO_ACKNOWLEDGE, and the handler's own onError defaults to a no-op), so this log line is the - * only evidence the failure ever happened - without it a failing trigger, rollup, posting or - * register lookup is indistinguishable from one that never fired. + * A handler that throws must leave a trace, and carry the throwable itself - logging only its + * message loses the stack trace, which is the whole diagnostic value. The failure is now also + * handed back to the broker (below), but that is a separate guarantee: redelivery re-runs the work, + * it does not explain it. Without this line a failing trigger, rollup, posting or register lookup + * is indistinguishable from one that never fired. */ @Test void aThrowingHandlerIsReportedInsteadOfSilentlyDiscarded() throws Exception { @@ -190,7 +202,8 @@ void aThrowingHandlerIsReportedInsteadOfSilentlyDiscarded() throws Exception { TextMessage message = mock(TextMessage.class); when(message.getText()).thenReturn("{}"); - capturedListener.onMessage(message); + // The failure escapes the listener now, so the report is asserted around the throw. + assertThrows(Exception.class, () -> capturedListener.onMessage(message)); List errors = appender.list.stream() .filter(event -> event.getLevel() == Level.ERROR) @@ -204,4 +217,50 @@ void aThrowingHandlerIsReportedInsteadOfSilentlyDiscarded() throws Exception { assertNotNull(error.getThrowableProxy(), "the throwable itself must be logged - passing only getMessage() loses the stack trace, " + "which is the whole diagnostic value"); } + + /** + * The failure must also LEAVE the listener, because that is the only thing the broker can observe. + * A swallowed exception is indistinguishable from success: the message is acknowledged and the + * event is gone for good, which also strands every handler whose correctness depends on a second + * delivery - the generated posting glue repairs a half-written post on redelivery. + */ + @Test + void aThrowingHandlerIsHandedBackToTheBrokerRatherThanAcknowledged() throws Exception { + when(tenantConfigurationService.resolveInjectableForCurrentTenant()).thenReturn(Map.of()); + handler.failWith = new IllegalStateException("journal line 3 of 5 violated a constraint"); + + TextMessage message = mock(TextMessage.class); + when(message.getText()).thenReturn("{}"); + + Exception thrown = assertThrows(Exception.class, () -> capturedListener.onMessage(message), + "the failure must escape onMessage - otherwise the broker sees a successful delivery"); + assertEquals("journal line 3 of 5 violated a constraint", thrown.getCause() + .getMessage(), + "the original failure must survive as the cause, not be flattened into a message"); + } + + /** + * Reporting still happens first - handing the message back must not skip the handler's own hook. + */ + @Test + void theHandlersOnErrorStillRunsBeforeTheMessageIsHandedBack() throws Exception { + when(tenantConfigurationService.resolveInjectableForCurrentTenant()).thenReturn(Map.of()); + handler.failWith = new IllegalStateException("boom"); + + TextMessage message = mock(TextMessage.class); + when(message.getText()).thenReturn("{}"); + + assertThrows(Exception.class, () -> capturedListener.onMessage(message)); + + assertEquals(List.of("boom"), handler.reportedErrors, "onError must still be invoked for the failed attempt"); + } + + /** + * Without a policy the broker falls back to its own defaults, so the two listener paths would + * disagree on how many times a failure is retried before it is dead-lettered. + */ + @Test + void theSubscriptionBoundsItsRetriesLikeTheJavaScriptListenerPath() { + verify(connectionFactory).configureRedeliveryPolicy(connection, queue); + } } diff --git a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/config/ActiveMQConnectionArtifactsFactory.java b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/config/ActiveMQConnectionArtifactsFactory.java index 3db32a659e8..fb405b76b2b 100644 --- a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/config/ActiveMQConnectionArtifactsFactory.java +++ b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/config/ActiveMQConnectionArtifactsFactory.java @@ -9,10 +9,15 @@ */ package org.eclipse.dirigible.components.listeners.config; +import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.RedeliveryPolicy; +import org.apache.activemq.broker.region.policy.RedeliveryPolicyMap; +import org.apache.activemq.command.ActiveMQDestination; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import jakarta.jms.Connection; +import jakarta.jms.Destination; import jakarta.jms.ExceptionListener; import jakarta.jms.JMSException; import jakarta.jms.Session; @@ -23,6 +28,15 @@ @Component public class ActiveMQConnectionArtifactsFactory { + /** The Constant INITIAL_REDELIVERY_DELAY. */ + private static final int INITIAL_REDELIVERY_DELAY = 1000; + + /** The Constant REDELIVERY_DELAY. */ + private static final int REDELIVERY_DELAY = 5000; + + /** The Constant MAXIMUM_REDELIVERIES. */ + private static final int MAXIMUM_REDELIVERIES = 3; + /** The connection factory. */ private final ActiveMQConnectionFactory connectionFactory; @@ -89,4 +103,31 @@ public Connection createConnection(ExceptionListener exceptionListener, String c } } + /** + * Bounds how often a failed delivery is retried before the broker gives up and dead-letters the + * message. The session is AUTO_ACKNOWLEDGE, which for an asynchronous listener acknowledges only + * after {@code onMessage} RETURNS - so a listener that throws leaves the message unacknowledged and + * this policy is what decides its fate. A listener that swallows its exception never reaches here: + * to the broker, swallowing and succeeding are the same outcome. + * + * Every listener path must apply this, so the retry budget is one number rather than one per + * caller. + * + * @param connection the connection + * @param destination the destination + */ + public void configureRedeliveryPolicy(Connection connection, Destination destination) { + if (connection instanceof ActiveMQConnection amqConnection && destination instanceof ActiveMQDestination amqDestination) { + RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy(); + + redeliveryPolicy.setInitialRedeliveryDelay(INITIAL_REDELIVERY_DELAY); + redeliveryPolicy.setRedeliveryDelay(REDELIVERY_DELAY); + redeliveryPolicy.setUseExponentialBackOff(true); + redeliveryPolicy.setMaximumRedeliveries(MAXIMUM_REDELIVERIES); + + RedeliveryPolicyMap policyMap = amqConnection.getRedeliveryPolicyMap(); + policyMap.put(amqDestination, redeliveryPolicy); + } + } + } diff --git a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenerManager.java b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenerManager.java index 294576328f1..72e757e1c14 100644 --- a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenerManager.java +++ b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenerManager.java @@ -11,10 +11,6 @@ import jakarta.jms.MessageConsumer; import jakarta.jms.*; -import org.apache.activemq.ActiveMQConnection; -import org.apache.activemq.RedeliveryPolicy; -import org.apache.activemq.broker.region.policy.RedeliveryPolicyMap; -import org.apache.activemq.command.ActiveMQDestination; import org.eclipse.dirigible.components.listeners.config.ActiveMQConnectionArtifactsFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -26,15 +22,6 @@ */ public class ListenerManager { - /** The Constant INITIAL_REDELIVERY_DELAY. */ - private static final int INITIAL_REDELIVERY_DELAY = 1000; - - /** The Constant REDELIVERY_DELAY. */ - private static final int REDELIVERY_DELAY = 5000; - - /** The Constant MAXIMUM_REDELIVERIES. */ - private static final int MAXIMUM_REDELIVERIES = 3; - /** The Constant LOGGER. */ private static final Logger LOGGER = LoggerFactory.getLogger(ListenerManager.class); @@ -83,7 +70,7 @@ public synchronized void startListener() { Session session = connectionArtifactsFactory.createSession(connection); Destination destination = createDestination(session); - configureRedeliveryPolicy(connection, destination); + connectionArtifactsFactory.configureRedeliveryPolicy(connection, destination); MessageConsumer consumer = session.createConsumer(destination); @@ -116,36 +103,6 @@ private Destination createDestination(Session session) throws JMSException { }; } - /** - * Configure redelivery policy. - * - * @param connection the connection - * @param destination the destination - */ - private void configureRedeliveryPolicy(Connection connection, Destination destination) { - if (connection instanceof ActiveMQConnection amqConnection && destination instanceof ActiveMQDestination amqDestination) { - RedeliveryPolicy redeliveryPolicy = createRedeliveryPolicy(); - RedeliveryPolicyMap policyMap = amqConnection.getRedeliveryPolicyMap(); - policyMap.put(amqDestination, redeliveryPolicy); - } - } - - /** - * Creates the redelivery policy. - * - * @return the redelivery policy - */ - private RedeliveryPolicy createRedeliveryPolicy() { - RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy(); - - redeliveryPolicy.setInitialRedeliveryDelay(INITIAL_REDELIVERY_DELAY); - redeliveryPolicy.setRedeliveryDelay(REDELIVERY_DELAY); - redeliveryPolicy.setUseExponentialBackOff(true); - redeliveryPolicy.setMaximumRedeliveries(MAXIMUM_REDELIVERIES); - - return redeliveryPolicy; - } - /** * Stop listener. */