mirror of https://github.com/apache/activemq.git
[AMQ-6587] additional contention window with new sub while gc is in progress - need to verify new sub is actually subscribed to the candidate for deletion. Fix and test
(cherry picked from commit 0ee4f5b843
)
This commit is contained in:
parent
c5315d6d5e
commit
2d4fb4cbab
|
@ -266,8 +266,12 @@ public abstract class AbstractRegion implements Region {
|
||||||
if (timeout == 0) {
|
if (timeout == 0) {
|
||||||
for (Iterator<Subscription> iter = subscriptions.values().iterator(); iter.hasNext();) {
|
for (Iterator<Subscription> iter = subscriptions.values().iterator(); iter.hasNext();) {
|
||||||
Subscription sub = iter.next();
|
Subscription sub = iter.next();
|
||||||
if (sub.matches(destination)) {
|
if (sub.matches(destination) ) {
|
||||||
throw new JMSException("Destination still has an active subscription: " + destination);
|
// may be a new sub created after gc decision, verify if really subscribed
|
||||||
|
Destination toDelete = destinations.get(destination);
|
||||||
|
if (toDelete != null && toDelete.getDestinationStatistics().getConsumers().getCount() > 0 ) {
|
||||||
|
throw new JMSException("Destination still has an active subscription: " + destination);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -148,4 +148,82 @@ public class DestinationGCStressTest {
|
||||||
assertFalse("failed on unexpected log event", failed.get());
|
assertFalse("failed on unexpected log event", failed.get());
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test(timeout = 60000)
|
||||||
|
public void testAddRemoveWildcardWithGc() throws Exception {
|
||||||
|
|
||||||
|
org.apache.log4j.Logger log4jLogger =
|
||||||
|
org.apache.log4j.Logger.getLogger(RegionBroker.class);
|
||||||
|
final AtomicBoolean failed = new AtomicBoolean(false);
|
||||||
|
|
||||||
|
Appender appender = new DefaultTestAppender() {
|
||||||
|
@Override
|
||||||
|
public void doAppend(LoggingEvent event) {
|
||||||
|
if (event.getLevel().equals(Level.ERROR) && event.getMessage().toString().startsWith("Failed to remove inactive")) {
|
||||||
|
logger.info("received unexpected log message: " + event.getMessage());
|
||||||
|
failed.set(true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
log4jLogger.addAppender(appender);
|
||||||
|
try {
|
||||||
|
|
||||||
|
final AtomicInteger max = new AtomicInteger(10000);
|
||||||
|
|
||||||
|
final ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?create=false");
|
||||||
|
factory.setWatchTopicAdvisories(false);
|
||||||
|
Connection connection = factory.createConnection();
|
||||||
|
connection.start();
|
||||||
|
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||||
|
|
||||||
|
ExecutorService executorService = Executors.newCachedThreadPool();
|
||||||
|
for (int i = 0; i < 1; i++) {
|
||||||
|
executorService.submit(new Runnable() {
|
||||||
|
@Override
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
Connection c = factory.createConnection();
|
||||||
|
c.start();
|
||||||
|
Session s = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||||
|
MessageProducer producer = s.createProducer(null);
|
||||||
|
Message message = s.createTextMessage();
|
||||||
|
int j;
|
||||||
|
while ((j = max.decrementAndGet()) > 0) {
|
||||||
|
producer.send(new ActiveMQTopic("A." + j), message);
|
||||||
|
}
|
||||||
|
} catch (Exception ignored) {
|
||||||
|
ignored.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
executorService.submit(new Runnable() {
|
||||||
|
@Override
|
||||||
|
public void run() {
|
||||||
|
for (int i = 0; i < 100; i++) {
|
||||||
|
try {
|
||||||
|
MessageConsumer messageConsumer = session.createConsumer(new ActiveMQTopic(">"));
|
||||||
|
messageConsumer.close();
|
||||||
|
|
||||||
|
} catch (Exception ignored) {
|
||||||
|
ignored.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
executorService.shutdown();
|
||||||
|
executorService.awaitTermination(60, TimeUnit.SECONDS);
|
||||||
|
|
||||||
|
logger.info("Done");
|
||||||
|
|
||||||
|
connection.close();
|
||||||
|
|
||||||
|
} finally {
|
||||||
|
log4jLogger.removeAppender(appender);
|
||||||
|
}
|
||||||
|
assertFalse("failed on unexpected log event", failed.get());
|
||||||
|
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue