ServerSessionImpl gets stuck if the WorkManager encounteres a RejectedExecutionException #2647

Activity

1. otbutz commented on Oct 12, 2026

   

   otbutz

   on Oct 12, 2026

   Author

   More actions

   This also affects ActiveMQEndpointWorker.connect()
2. otbutz

   changed the title [-]ServerSessionImpl gets stuck if the WorkManagers encounteres a RejectedExecutionException[/-] [+]ServerSessionImpl gets stuck if the WorkManager encounteres a RejectedExecutionException[/+] on Oct 13, 2026
3. otbutz commented on Oct 14, 2026

   

   otbutz

   on Oct 14, 2026

   Author

   More actions

   Sadly #2648 doesn't fix the issue as confirmed by @rzo1 in apache/tomee#3090
4. rzo1 commented on Oct 14, 2026

   

   rzo1

   on Oct 14, 2026

   More actions

   I can confirm this, and #2648 alone isn't enough. Widening the catch in ServerSessionImpl.start() gets the session out of its stuck "running" state. But the JMSException then reaches the endpoint worker's ExceptionListener, which disconnects and calls connect(). connect() schedules the reconnect on the same work manager, which is still saturated, so that is rejected too. Then:

   } catch (WorkException e) {
       running = false;
       LOG.error("Work Manager did not accept work: ", e);
   }

   running = false means the endpoint never tries to connect again, so the consumer is gone for good. This is the sequence from a run against current main:

   WARN  ServerSessionImpl      - Failed to schedule work for session ServerSessionImpl:{...}, marking not running
   ERROR ActiveMQEndpointWorker - Connection to broker failed: Start failed: jakarta.resource.spi.work.WorkRejectedException: thread pool saturated
   ERROR ActiveMQEndpointWorker - Work Manager did not accept work:
   

   Here is a test for activemq-ra that reproduces it. Its work manager throws WorkRejectedException while a flag is set, which is what a saturated GeronimoWorkManager does with GERONIMO-6911 applied. It doesn't depend on #2648 and fails on main today:

   RejectedWorkRecoveryTest.java

   /**
    * Licensed to the Apache Software Foundation (ASF) under one or more
    * contributor license agreements.  See the NOTICE file distributed with
    * this work for additional information regarding copyright ownership.
    * The ASF licenses this file to You under the Apache License, Version 2.0
    * (the "License"); you may not use this file except in compliance with
    * the License.  You may obtain a copy of the License at
    *
    *      http://www.apache.org/licenses/LICENSE-2.0
    *
    * Unless required by applicable law or agreed to in writing, software
    * distributed under the License is distributed on an "AS IS" BASIS,
    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    * See the License for the specific language governing permissions and
    * limitations under the License.
    */
   package org.apache.activemq.ra;
   
   import static org.junit.Assert.assertTrue;
   
   import java.lang.reflect.Method;
   import java.util.Timer;
   import java.util.concurrent.CountDownLatch;
   import java.util.concurrent.TimeUnit;
   import java.util.concurrent.atomic.AtomicBoolean;
   
   import jakarta.jms.Connection;
   import jakarta.jms.Message;
   import jakarta.jms.MessageListener;
   import jakarta.jms.MessageProducer;
   import jakarta.jms.Queue;
   import jakarta.jms.Session;
   import jakarta.jms.TextMessage;
   import jakarta.resource.spi.BootstrapContext;
   import jakarta.resource.spi.UnavailableException;
   import jakarta.resource.spi.XATerminator;
   import jakarta.resource.spi.endpoint.MessageEndpoint;
   import jakarta.resource.spi.endpoint.MessageEndpointFactory;
   import jakarta.resource.spi.work.ExecutionContext;
   import jakarta.resource.spi.work.Work;
   import jakarta.resource.spi.work.WorkContext;
   import jakarta.resource.spi.work.WorkException;
   import jakarta.resource.spi.work.WorkListener;
   import jakarta.resource.spi.work.WorkManager;
   import jakarta.resource.spi.work.WorkRejectedException;
   import jakarta.transaction.TransactionSynchronizationRegistry;
   
   import javax.transaction.xa.XAResource;
   
   import org.apache.activemq.ActiveMQConnectionFactory;
   import org.apache.activemq.command.ActiveMQQueue;
   import org.junit.Test;
   
   /**
    * Rejects the work of the work manager for a while, as a saturated thread pool does,
    * and checks that the endpoint consumes again once work is accepted.
    */
   public class RejectedWorkRecoveryTest {
   
       private final AtomicBoolean reject = new AtomicBoolean();
   
       @Test(timeout = 60000)
       public void testEndpointRecoversAfterRejectedWork() throws Exception {
           ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
           Connection connection = factory.createConnection();
           connection.start();
           Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
           MessageProducer producer = session.createProducer(new ActiveMQQueue("TEST"));
   
           ActiveMQResourceAdapter adapter = new ActiveMQResourceAdapter();
           adapter.setServerUrl("vm://localhost?broker.persistent=false&create=false");
           adapter.start(new RejectingBootstrapContext());
   
           final CountDownLatch recovered = new CountDownLatch(1);
           final CountDownLatch connected = new CountDownLatch(1);
           MessageEndpointFactory endpointFactory = new EndpointFactory(new MessageListener() {
               @Override
               public void onMessage(Message message) {
                   try {
                       if ("ping".equals(((TextMessage) message).getText())) {
                           connected.countDown();
                       } else if ("after".equals(((TextMessage) message).getText())) {
                           recovered.countDown();
                       }
                   } catch (Exception e) {
                       throw new IllegalStateException(e);
                   }
               }
           });
   
           ActiveMQActivationSpec activationSpec = new ActiveMQActivationSpec();
           activationSpec.setDestinationType(Queue.class.getName());
           activationSpec.setDestination("TEST");
           activationSpec.setMaxSessions("1");
           activationSpec.setResourceAdapter(adapter);
           activationSpec.validate();
           adapter.endpointActivation(endpointFactory, activationSpec);
   
           producer.send(session.createTextMessage("ping"));
           assertTrue("endpoint did not consume before the rejection", connected.await(10, TimeUnit.SECONDS));
   
           // the thread pool is saturated, the work for this message is rejected
           reject.set(true);
           producer.send(session.createTextMessage("rejected"));
           Thread.sleep(2000);
   
           // the thread pool recovers
           reject.set(false);
           producer.send(session.createTextMessage("after"));
           assertTrue("endpoint never consumed again after rejected work", recovered.await(20, TimeUnit.SECONDS));
   
           connection.close();
           adapter.endpointDeactivation(endpointFactory, activationSpec);
           adapter.stop();
       }
   
       private abstract static class ListenerAdapter implements MessageEndpoint, MessageListener {
           @Override
           public void beforeDelivery(Method method) {
           }
   
           @Override
           public void afterDelivery() {
           }
   
           @Override
           public void release() {
           }
       }
   
       private static final class EndpointFactory implements MessageEndpointFactory {
           private final MessageListener listener;
   
           private EndpointFactory(MessageListener listener) {
               this.listener = listener;
           }
   
           @Override
           public MessageEndpoint createEndpoint(XAResource resource) throws UnavailableException {
               return new ListenerAdapter() {
                   @Override
                   public void onMessage(Message message) {
                       listener.onMessage(message);
                   }
               };
           }
   
           @Override
           public MessageEndpoint createEndpoint(XAResource resource, long timeout) throws UnavailableException {
               return createEndpoint(resource);
           }
   
           @Override
           public boolean isDeliveryTransacted(Method method) {
               return false;
           }
   
           @Override
           public String getActivationName() {
               return null;
           }
   
           @Override
           public Class<?> getEndpointClass() {
               return null;
           }
       }
   
       private final class RejectingBootstrapContext implements BootstrapContext {
           @Override
           public WorkManager getWorkManager() {
               return new WorkManager() {
                   @Override
                   public void doWork(Work work) throws WorkException {
                       run(work);
                   }
   
                   @Override
                   public void doWork(Work work, long timeout, ExecutionContext context, WorkListener listener) throws WorkException {
                       run(work);
                   }
   
                   @Override
                   public long startWork(Work work) throws WorkException {
                       run(work);
                       return 0;
                   }
   
                   @Override
                   public long startWork(Work work, long timeout, ExecutionContext context, WorkListener listener) throws WorkException {
                       run(work);
                       return 0;
                   }
   
                   @Override
                   public void scheduleWork(Work work) throws WorkException {
                       run(work);
                   }
   
                   @Override
                   public void scheduleWork(Work work, long timeout, ExecutionContext context, WorkListener listener) throws WorkException {
                       run(work);
                   }
   
                   private void run(Work work) throws WorkException {
                       if (reject.get()) {
                           throw new WorkRejectedException("thread pool saturated");
                       }
                       new Thread(work).start();
                   }
               };
           }
   
           @Override
           public XATerminator getXATerminator() {
               return null;
           }
   
           @Override
           public Timer createTimer() {
               return null;
           }
   
           @Override
           public boolean isContextSupported(Class<? extends WorkContext> workContextClass) {
               return false;
           }
   
           @Override
           public TransactionSynchronizationRegistry getTransactionSynchronizationRegistry() {
               return null;
           }
       }
   }

   A rejection like this is temporary, so I don't think connect() should give up when the work manager rejects the reconnect. It could:

   • treat WorkRejectedException as transient and retry with the same back-off that pause() uses for broker connection failures. The retry could be scheduled through the Timer from BootstrapContext.createTimer(), so the RA doesn't create threads of its own;
   • keep running = false only for work exceptions that really are permanent, or when the endpoint is being stopped.

   The TomEE side of the reproducer is in apache/tomee#3090. It shows both the unpatched case (RejectedExecutionException) and the case with GERONIMO-6911 applied.
5. otbutz commented on Oct 14, 2026

   

   otbutz

   on Oct 14, 2026

   Author

   More actions

   A rejection like this is temporary, so I don't think connect() should give up when the work manager rejects the reconnect.

   Maybe we should use jakarta.resource.spi.work.RetryableWorkRejectedException instead?