rzo1 commented on issue #2647:
URL: https://github.com/apache/activemq/issues/2647#issuecomment-6165610414
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:
```java
} 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:
<details>
<summary>RejectedWorkRecoveryTest.java</summary>
```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;
}
}
}
```
</details>
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
