Failed client-Java listener delivery is retried, not lost

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
 

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)
 

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);
         }
     }
 

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<String, String> observedDuringDispatch;
         RuntimeException failWith;
+        final List<String> reportedErrors = new ArrayList<>();
 
         @Override
         public String destination() {
@@ -77,19 +80,27 @@ 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<ILoggingEvent> appender;
     private Logger consumerLogger;
 
     @BeforeEach
     @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<ILoggingEvent> 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);
+    }
 }

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);
+        }
+    }
+
 }