This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch activemq-5.19.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.19.x by this push:
new 9d5f42d9f0 [AMQ-9692] Support destination gc sweep of destinations
with only wildcard consumers (#1484) (#2509)
9d5f42d9f0 is described below
commit 9d5f42d9f0c7f7a01631962d87b8f632d6638a76
Author: Matt Pavlovich <[email protected]>
AuthorDate: Wed Sep 2 12:20:59 2026 -0500
[AMQ-9692] Support destination gc sweep of destinations with only wildcard
consumers (#1484) (#2509)
- 5.19.x backport fixes
(cherry picked from commit bbe46065d0253935d9c3629975372efe119246ec)
---
.../activemq/broker/region/AbstractRegion.java | 14 +-
.../activemq/broker/region/BaseDestination.java | 82 +++--
.../apache/activemq/broker/region/Destination.java | 1 +
.../activemq/broker/region/DestinationFilter.java | 5 +
.../apache/activemq/broker/region/TempQueue.java | 3 +-
.../activemq/broker/region/TempQueueRegion.java | 2 +-
.../activemq/broker/region/policy/PolicyEntry.java | 23 ++
.../apache/activemq/java/JavaPolicyEntryTest.java | 48 +--
.../region/DestinationCanGcConsumerTest.java | 205 ++++++++++++
.../activemq/broker/region/DestinationGCTest.java | 305 +++++++++++++----
.../DestinationGCWildcardDurableSubTest.java | 360 +++++++++++++++++++++
.../broker/region/DestinationIsActiveTest.java | 341 +++++++++++++++++++
12 files changed, 1273 insertions(+), 116 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
index e62bfed81e..7233b1800e 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
@@ -260,16 +260,16 @@ public abstract class AbstractRegion implements Region {
}
@Override
- public void removeDestination(ConnectionContext context,
ActiveMQDestination destination, long timeout)
- throws Exception {
-
+ public void removeDestination(ConnectionContext context,
ActiveMQDestination destination, long timeout) throws Exception {
// No timeout.. then try to shut down right way, fails if there are
// current subscribers.
if (timeout == 0) {
- for (Iterator<Subscription> iter =
subscriptions.values().iterator(); iter.hasNext();) {
- Subscription sub = iter.next();
- if (sub.matches(destination) ) {
- throw new JMSException("Destination: " + destination + "
still has an active subscription: " + sub);
+ final var dest = destinations.get(destination);
+ if (dest != null) {
+ for (var sub : subscriptions.values()) {
+ if (sub.matches(destination) && dest.isActive()) {
+ throw new JMSException("Destination: " + destination +
" still has an active subscription: " + sub);
+ }
}
}
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
index 192d1d0fcb..5dbf7cc337 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
@@ -19,6 +19,7 @@ package org.apache.activemq.broker.region;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
import javax.jms.ResourceAllocationException;
@@ -105,7 +106,8 @@ public abstract class BaseDestination implements
Destination {
private long inactiveTimeoutBeforeGC = DEFAULT_INACTIVE_TIMEOUT_BEFORE_GC;
private boolean gcIfInactive;
private boolean gcWithNetworkConsumers;
- private long lastActiveTime=0l;
+ private boolean gcWithOnlyWildcardConsumers;
+ private long lastActiveTime = 0L;
private boolean reduceMemoryFootprint = false;
protected final Scheduler scheduler;
private boolean disposed = false;
@@ -311,12 +313,42 @@ public abstract class BaseDestination implements
Destination {
@Override
public boolean isActive() {
- boolean isActive = destinationStatistics.getConsumers().getCount() > 0
||
- destinationStatistics.getProducers().getCount() > 0;
- if (isActive && isGcWithNetworkConsumers() &&
destinationStatistics.getConsumers().getCount() > 0) {
- isActive = hasRegularConsumers(getConsumers());
+ // if we have producers then we are active
+ if (destinationStatistics.getProducers().getCount() > 0) {
+ return true;
}
- return isActive;
+
+ // Check if we have active consumers that should prevent GC
+ if (destinationStatistics.getConsumers().getCount() > 0) {
+ // if we have consumers and both gcWithNetwork and gcOnlyWildcard
consumers
+ // are false we can just return true, otherwise we need to check
each consumer
+ return (!isGcWithNetworkConsumers() &&
!isGcWithOnlyWildcardConsumers()) ||
+ hasActiveConsumers();
+ }
+
+ return false;
+ }
+
+ Predicate<Subscription> canGcConsumer = subscription -> {
+ // if isGcWithNetworkConsumers() is true and this is a network
subscription then we can GC
+ boolean canGcNetwork = isGcWithNetworkConsumers() &&
subscription.getConsumerInfo().isNetworkSubscription();
+ // if isGcWithOnlyWildcardConsumers() is true and this is a
non-durable wildcard then we can GC.
+ // An attached durable subscription never permits gc - its
registration and pending messages
+ // live in the destination's store, which gc destroys. Note offline
durable subscriptions
+ // stay attached only with keepDurableSubsActive=true (the default);
brokers running with
+ // keepDurableSubsActive=false forfeit this protection while the
subscriber is offline.
+ return canGcNetwork || (isGcWithOnlyWildcardConsumers() &&
subscription.isWildcard()
+ && !subscription.getConsumerInfo().isDurable());
+ };
+
+ protected boolean hasActiveConsumers() {
+ final List<Subscription> consumers = getConsumers();
+ for (Subscription subscription: consumers) {
+ if (!canGcConsumer.test(subscription)) {
+ return true;
+ }
+ }
+ return false;
}
@Override
@@ -796,19 +828,36 @@ public abstract class BaseDestination implements
Destination {
return gcWithNetworkConsumers;
}
+ /**
+ * Indicate if it is ok to gc destinations that have only wildcard
consumers
+ * @param gcWithOnlyWildcardConsumers
+ */
+ public void setGcWithOnlyWildcardConsumers(boolean
gcWithOnlyWildcardConsumers) {
+ this.gcWithOnlyWildcardConsumers = gcWithOnlyWildcardConsumers;
+ }
+
+ public boolean isGcWithOnlyWildcardConsumers() {
+ return gcWithOnlyWildcardConsumers;
+ }
+
@Override
public void markForGC(long timeStamp) {
- if (isGcIfInactive() && this.lastActiveTime == 0 && isActive() == false
- && destinationStatistics.getMessages().getCount() == 0 &&
getInactiveTimeoutBeforeGC() > 0l) {
+ if (isGcIfInactive()
+ && this.lastActiveTime == 0
+ && destinationStatistics.getMessages().getCount() == 0
+ && getInactiveTimeoutBeforeGC() > 0L
+ && !isActive()) {
this.lastActiveTime = timeStamp;
}
}
@Override
public boolean canGC() {
- boolean result = false;
- final long currentLastActiveTime = this.lastActiveTime;
- if (isGcIfInactive() && currentLastActiveTime != 0l &&
destinationStatistics.getMessages().getCount() == 0L ) {
+ var result = false;
+ final var currentLastActiveTime = this.lastActiveTime;
+ if (isGcIfInactive()
+ && currentLastActiveTime != 0L
+ && destinationStatistics.getMessages().getCount() == 0L) {
if ((System.currentTimeMillis() - currentLastActiveTime) >=
getInactiveTimeoutBeforeGC()) {
result = true;
}
@@ -865,17 +914,6 @@ public abstract class BaseDestination implements
Destination {
@Override
public abstract List<Subscription> getConsumers();
- protected boolean hasRegularConsumers(List<Subscription> consumers) {
- boolean hasRegularConsumers = false;
- for (Subscription subscription: consumers) {
- if (!subscription.getConsumerInfo().isNetworkSubscription()) {
- hasRegularConsumers = true;
- break;
- }
- }
- return hasRegularConsumers;
- }
-
public ConnectionContext createConnectionContext() {
ConnectionContext answer = new ConnectionContext();
answer.setBroker(this.broker);
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Destination.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Destination.java
index 22ba14894b..2901fd0e18 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Destination.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Destination.java
@@ -267,4 +267,5 @@ public interface Destination extends Service, Task,
Message.MessageDestination {
void setAdvancedMessageStatisticsEnabled(boolean
advancedMessageStatisticsEnabled);
+ boolean isGcWithOnlyWildcardConsumers();
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
index 154f013d6c..59bcc86877 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
@@ -429,6 +429,11 @@ public class DestinationFilter implements Destination {
next.setAdvancedMessageStatisticsEnabled(advancedMessageStatisticsEnabled);
}
+ @Override
+ public boolean isGcWithOnlyWildcardConsumers() {
+ return next.isGcWithOnlyWildcardConsumers();
+ }
+
public void deleteSubscription(ConnectionContext context, SubscriptionKey
key) throws Exception {
if (next instanceof DestinationFilter) {
DestinationFilter filter = (DestinationFilter) next;
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueue.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueue.java
index be1daf76b1..f7c2126cc0 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueue.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueue.java
@@ -34,8 +34,7 @@ import org.slf4j.LoggerFactory;
*
*
*/
-public class TempQueue extends Queue implements TempDestination{
-
+public class TempQueue extends Queue implements TempDestination {
private static final Logger LOG = LoggerFactory.getLogger(TempQueue.class);
private final ActiveMQTempDestination tempDest;
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueueRegion.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueueRegion.java
index 78d695557f..0e431cafb0 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueueRegion.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TempQueueRegion.java
@@ -58,7 +58,7 @@ public class TempQueueRegion extends AbstractTempRegion {
super.removeDestination(context, destination, timeout);
}
-
+
/*
* For a Queue, dispatch order is imperative to match acks, so the
dispatch is deferred till
* the notification to ensure that the subscription chosen by the master
is used.
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
index 235ab7d5dc..373becfbb1 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
@@ -100,6 +100,7 @@ public class PolicyEntry extends DestinationMapEntry {
private boolean prioritizedMessages;
private boolean allConsumersExclusiveByDefault;
private boolean gcInactiveDestinations;
+ private boolean gcWithOnlyWildcardConsumers;
private boolean gcWithNetworkConsumers;
private long inactiveTimeoutBeforeGC =
BaseDestination.DEFAULT_INACTIVE_TIMEOUT_BEFORE_GC;
private boolean reduceMemoryFootprint;
@@ -265,6 +266,9 @@ public class PolicyEntry extends DestinationMapEntry {
if (isUpdate("gcInactiveDestinations", includedProperties)) {
destination.setGcIfInactive(isGcInactiveDestinations());
}
+ if (isUpdate("gcWithOnlyWildcardConsumers", includedProperties)) {
+
destination.setGcWithOnlyWildcardConsumers(isGcWithOnlyWildcardConsumers());
+ }
if (isUpdate("gcWithNetworkConsumers", includedProperties)) {
destination.setGcWithNetworkConsumers(isGcWithNetworkConsumers());
}
@@ -1090,6 +1094,25 @@ public class PolicyEntry extends DestinationMapEntry {
this.inactiveTimeoutBeforeGC = inactiveTimeoutBeforeGC;
}
+ /**
+ * Allow gc of inactive destinations whose only consumers are wildcard
subscriptions,
+ * supporting one-time-use destination patterns where a wildcard consumer
stays
+ * connected across destinations being created, drained and collected.
+ *
+ * An attached durable topic subscription always prevents gc - its
registration and
+ * pending messages live in the destination's store, which gc destroys.
Offline durable
+ * subscriptions remain attached only when the broker runs with
keepDurableSubsActive=true
+ * (the default); with keepDurableSubsActive=false this protection does
not apply while
+ * the subscriber is offline.
+ */
+ public void setGcWithOnlyWildcardConsumers(boolean
gcWithOnlyWildcardConsumers) {
+ this.gcWithOnlyWildcardConsumers = gcWithOnlyWildcardConsumers;
+ }
+
+ public boolean isGcWithOnlyWildcardConsumers() {
+ return gcWithOnlyWildcardConsumers;
+ }
+
public void setGcWithNetworkConsumers(boolean gcWithNetworkConsumers) {
this.gcWithNetworkConsumers = gcWithNetworkConsumers;
}
diff --git
a/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
b/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
index f388b69cad..a6bd1ce7bd 100644
---
a/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
+++
b/activemq-runtime-config/src/test/java/org/apache/activemq/java/JavaPolicyEntryTest.java
@@ -662,7 +662,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
//initial config
setAllDestPolicyProperties(entry, true, true, 10,
- 100, 200, 1000, 400, 40, 30, true, true, 1000, true, true,
+ 100, 200, 1000, 400, 40, 30, true, true, true, 1000, true,
true,
30, true, true, true, true, true, true, true, true, true);
setAllQueuePolicyProperties(entry, 10000, true, true, true, true, 100,
100, true, true);
@@ -675,7 +675,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
//validate config
assertAllDestPolicyProperties(getQueue("Before"), true, true, 10,
- 100, 200, 1000, 400, 40, 30, true, true, 1000, true, true,
+ 100, 200, 1000, 400, 40, 30, true, true, true, 1000, true,
true,
30, true, true, true,true, true, true, true, true, true);
assertAllQueuePolicyProperties(getQueue("Before"), 10000, true, true,
true, true, 100,
100, true, true);
@@ -683,7 +683,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
//change config
setAllDestPolicyProperties(entry, false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false,false, false, false, false, false,
false);
setAllQueuePolicyProperties(entry, 100000, false, false, false, false,
1000,
1000, false, false);
@@ -692,14 +692,14 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
TimeUnit.SECONDS.sleep(SLEEP);
assertAllDestPolicyProperties(getQueue("Before"), false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false,false, false, false, false, false,
false);
assertAllQueuePolicyProperties(getQueue("Before"), 100000, false,
false, false, false, 1000,
1000, false, false);
//check new dest
assertAllDestPolicyProperties(getQueue("After"), false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false, false, false, false, false, false,
false);
assertAllQueuePolicyProperties(getQueue("After"), 100000, false,
false, false, false, 1000,
1000, false, false);
@@ -713,7 +713,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
//initial config
setAllDestPolicyProperties(entry, true, true, 10,
- 100, 200, 1000, 400, 40, 30, true, true, 1000, true, true,
+ 100, 200, 1000, 400, 40, 30, true, true, true, 1000, true,
true,
30, true, true, true, true, true, true, true, true, true);
setAllTopicPolicyProperties(entry, 10000, true);
@@ -725,14 +725,14 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
//validate config
assertAllDestPolicyProperties(getTopic("Before"), true, true, 10,
- 100, 200, 1000, 400, 40, 30, true, true, 1000, true, true,
+ 100, 200, 1000, 400, 40, 30, true, true, true, 1000, true,
true,
30, true, true, true, true, true, true, true, true, true);
assertAllTopicPolicyProperties(getTopic("Before"), 10000, true);
//change config
setAllDestPolicyProperties(entry, false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false, false, false, false, false, false,
false);
setAllTopicPolicyProperties(entry, 100000, false);
@@ -740,13 +740,13 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
TimeUnit.SECONDS.sleep(SLEEP);
assertAllDestPolicyProperties(getTopic("Before"), false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false, false, false, false, false, false,
false);
assertAllTopicPolicyProperties(getTopic("Before"), 100000, false);
//check new dest
assertAllDestPolicyProperties(getTopic("After"), false, false, 100,
- 1000, 2000, 10000, 4000, 400, 300, false, false, 1000, false,
false,
+ 1000, 2000, 10000, 4000, 400, 300, false, false, false, 1000,
false, false,
300, false, false, false, false, false, false, false, false,
false);
assertAllTopicPolicyProperties(getTopic("After"), 100000, false);
}
@@ -820,6 +820,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
properties.add("cursorMemoryHighWaterMark");
properties.add("storeUsageHighWaterMark");
properties.add("gcInactiveDestinations");
+ properties.add("gcWithOnlyWildcardConsumers");
properties.add("gcWithNetworkConsumers");
properties.add("inactiveTimeoutBeforeGC");
properties.add("reduceMemoryFootprint");
@@ -862,12 +863,12 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
private void setAllDestPolicyProperties(PolicyEntry entry, boolean
producerFlowControl,
boolean alwaysRetroactive, long blockedProducerWarningInterval,
int maxPageSize,
int maxBrowsePageSize, long minimumMessageSize, int
maxExpirePageSize, int cursorMemoryHighWaterMark,
- int storeUsageHighWaterMark, boolean gcInactiveDestinations,
boolean gcWithNetworkConsumers,
- long inactiveTimeoutBeforeGC,boolean reduceMemoryFootprint,
boolean doOptimizeMessageStore,
- int optimizeMessageStoreInFlightLimit, boolean
advisoryForConsumed, boolean advisoryForDelivery,
- boolean advisoryForDispatched, boolean
advisoryForDiscardingMessages, boolean advisoryForSlowConsumers,
- boolean advisoryForFastProducers, boolean advisoryWhenFull,
boolean includeBodyForAdvisory,
- boolean sendAdvisoryIfNoConsumers) {
+ int storeUsageHighWaterMark, boolean gcInactiveDestinations,
boolean gcWithOnlyWildcardConsumers,
+ boolean gcWithNetworkConsumers, long inactiveTimeoutBeforeGC,
boolean reduceMemoryFootprint,
+ boolean doOptimizeMessageStore, int
optimizeMessageStoreInFlightLimit, boolean advisoryForConsumed,
+ boolean advisoryForDelivery, boolean advisoryForDispatched,
boolean advisoryForDiscardingMessages,
+ boolean advisoryForSlowConsumers, boolean
advisoryForFastProducers, boolean advisoryWhenFull,
+ boolean includeBodyForAdvisory, boolean sendAdvisoryIfNoConsumers)
{
entry.setProducerFlowControl(producerFlowControl);
entry.setAlwaysRetroactive(alwaysRetroactive);
@@ -879,6 +880,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
entry.setCursorMemoryHighWaterMark(cursorMemoryHighWaterMark);
entry.setStoreUsageHighWaterMark(storeUsageHighWaterMark);
entry.setGcInactiveDestinations(gcInactiveDestinations);
+ entry.setGcWithOnlyWildcardConsumers(gcWithOnlyWildcardConsumers);
entry.setGcWithNetworkConsumers(gcWithNetworkConsumers);
entry.setInactiveTimeoutBeforeGC(inactiveTimeoutBeforeGC);
entry.setReduceMemoryFootprint(reduceMemoryFootprint);
@@ -920,13 +922,12 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
private void assertAllDestPolicyProperties(BaseDestination dest, boolean
producerFlowControl,
boolean alwaysRetroactive, long blockedProducerWarningInterval,
int maxPageSize,
int maxBrowsePageSize, long minimumMessageSize, int
maxExpirePageSize, int cursorMemoryHighWaterMark,
- int storeUsageHighWaterMark, boolean gcInactiveDestinations,
boolean gcWithNetworkConsumers,
- long inactiveTimeoutBeforeGC,boolean reduceMemoryFootprint,
boolean doOptimizeMessageStore,
- int optimizeMessageStoreInFlightLimit, boolean
advisoryForConsumed, boolean advisoryForDelivery,
- boolean advisoryForDispatched, boolean
advisoryForDiscardingMessages, boolean advisoryForSlowConsumers,
- boolean advisoryForFastProducers, boolean advisoryWhenFull,
boolean includeBodyForAdvisory,
- boolean sendAdvisoryIfNoConsumers) {
-
+ int storeUsageHighWaterMark, boolean gcInactiveDestinations,
boolean gcWithOnlyWildcardConsumers,
+ boolean gcWithNetworkConsumers, long inactiveTimeoutBeforeGC,
boolean reduceMemoryFootprint,
+ boolean doOptimizeMessageStore, int
optimizeMessageStoreInFlightLimit, boolean advisoryForConsumed,
+ boolean advisoryForDelivery, boolean advisoryForDispatched,
boolean advisoryForDiscardingMessages,
+ boolean advisoryForSlowConsumers, boolean
advisoryForFastProducers, boolean advisoryWhenFull,
+ boolean includeBodyForAdvisory, boolean sendAdvisoryIfNoConsumers)
{
assertEquals(producerFlowControl, dest.isProducerFlowControl());
assertEquals(alwaysRetroactive, dest.isAlwaysRetroactive());
@@ -938,6 +939,7 @@ public class JavaPolicyEntryTest extends
RuntimeConfigTestSupport {
assertEquals(cursorMemoryHighWaterMark,
dest.getCursorMemoryHighWaterMark());
assertEquals(storeUsageHighWaterMark,
dest.getStoreUsageHighWaterMark());
assertEquals(gcInactiveDestinations, dest.isGcIfInactive());
+ assertEquals(gcWithOnlyWildcardConsumers,
dest.isGcWithOnlyWildcardConsumers());
assertEquals(gcWithNetworkConsumers, dest.isGcWithNetworkConsumers());
assertEquals(inactiveTimeoutBeforeGC,
dest.getInactiveTimeoutBeforeGC());
assertEquals(reduceMemoryFootprint, dest.isReduceMemoryFootprint());
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationCanGcConsumerTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationCanGcConsumerTest.java
new file mode 100644
index 0000000000..9590490e2c
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationCanGcConsumerTest.java
@@ -0,0 +1,205 @@
+/**
+ * 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.broker.region;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ConsumerInfo;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+/**
+ * [AMQ-9692] Direct coverage of the package-private
BaseDestination.canGcConsumer
+ * predicate, which decides whether a single subscription may be ignored by
+ * destination gc:
+ *
+ * (gcWithNetworkConsumers AND network) OR
+ * (gcWithOnlyWildcardConsumers AND wildcard AND NOT durable)
+ *
+ * The table covers every combination of the two policy flags and the three
+ * subscription traits. Notable rows it pins:
+ * - a durable subscription is never gc-eligible via the wildcard branch,
+ * regardless of flags
+ * - the network branch predates the durable exclusion and intentionally does
+ * not consult isDurable() - a durable network (bridge demand) subscription
+ * remains gc-eligible under gcWithNetworkConsumers, as before this feature
+ *
+ * gcNC = gcWithNetworkConsumers, gcWC = gcWithOnlyWildcardConsumers
+ * net = network subscription, wild = wildcard, dur = durable
+ */
+@RunWith(Parameterized.class)
+public class DestinationCanGcConsumerTest {
+
+ private static BrokerService brokerService;
+ private static BaseDestination baseDestination;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ brokerService = new BrokerService();
+ brokerService.setPersistent(false);
+ brokerService.setUseJmx(false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+
+ var queue = new ActiveMQQueue("amq.gc.canGcConsumer");
+ brokerService.getBroker().addDestination(
+ brokerService.getAdminConnectionContext(), queue, false);
+ baseDestination = (BaseDestination)
brokerService.getDestination(queue);
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ if (brokerService != null) {
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ }
+ }
+
+ @Parameterized.Parameters(name = "gcNC={0} gcWC={1} net={2} wild={3}
dur={4} exp={5}")
+ public static Collection<Object[]> data() {
+ var truthTable = new Object[][] {
+ // Plain app subscription - never gc-eligible
+ { false, false, false, false, false, false },
+ { false, true, false, false, false, false },
+ { true, false, false, false, false, false },
+ { true, true, false, false, false, false },
+
+ // Network subscription - gc-eligible iff gcNC
+ { false, false, true, false, false, false },
+ { false, true, true, false, false, false },
+ { true, false, true, false, false, true },
+ { true, true, true, false, false, true },
+
+ // Wildcard subscription - gc-eligible iff gcWC
+ { false, false, false, true, false, false },
+ { false, true, false, true, false, true },
+ { true, false, false, true, false, false },
+ { true, true, false, true, false, true },
+
+ // Durable (non-wildcard) subscription - never gc-eligible
+ { false, false, false, false, true, false },
+ { false, true, false, false, true, false },
+ { true, false, false, false, true, false },
+ { true, true, false, false, true, false },
+
+ // Durable wildcard subscription - the durable exclusion blocks
+ // the wildcard branch under every flag combination
+ { false, false, false, true, true, false },
+ { false, true, false, true, true, false },
+ { true, false, false, true, true, false },
+ { true, true, false, true, true, false },
+
+ // Network + wildcard - eligible via either enabled branch
+ { false, false, true, true, false, false },
+ { false, true, true, true, false, true },
+ { true, false, true, true, false, true },
+ { true, true, true, true, false, true },
+
+ // Network + durable - the network branch does not consult
+ // isDurable() (pre-existing gcWithNetworkConsumers semantics)
+ { false, false, true, false, true, false },
+ { false, true, true, false, true, false },
+ { true, false, true, false, true, true },
+ { true, true, true, false, true, true },
+
+ // Network + durable wildcard - eligible only via the network
+ // branch; the wildcard branch stays blocked by the durable
+ { false, false, true, true, true, false },
+ { false, true, true, true, true, false },
+ { true, false, true, true, true, true },
+ { true, true, true, true, true, true }
+ };
+ var params = new ArrayList<Object[]>();
+ for (var row : truthTable) {
+ params.add(row);
+ }
+ return params;
+ }
+
+ private final boolean gcWithNetworkConsumersEnabled;
+ private final boolean gcWithOnlyWildcardConsumersEnabled;
+ private final boolean networkSubscription;
+ private final boolean wildcardSubscription;
+ private final boolean durableSubscription;
+ private final boolean canGcExpected;
+
+ public DestinationCanGcConsumerTest(boolean gcWithNetworkConsumersEnabled,
boolean gcWithOnlyWildcardConsumersEnabled, boolean networkSubscription,
boolean wildcardSubscription, boolean durableSubscription, boolean
canGcExpected) {
+ this.gcWithNetworkConsumersEnabled = gcWithNetworkConsumersEnabled;
+ this.gcWithOnlyWildcardConsumersEnabled =
gcWithOnlyWildcardConsumersEnabled;
+ this.networkSubscription = networkSubscription;
+ this.wildcardSubscription = wildcardSubscription;
+ this.durableSubscription = durableSubscription;
+ this.canGcExpected = canGcExpected;
+ }
+
+ @Test
+ public void testCanGcConsumer() throws Exception {
+
baseDestination.setGcWithNetworkConsumers(gcWithNetworkConsumersEnabled);
+
baseDestination.setGcWithOnlyWildcardConsumers(gcWithOnlyWildcardConsumersEnabled);
+
+ var subscription = new
MockSubscription(baseDestination.getActiveMQDestination(),
+ networkSubscription, wildcardSubscription,
durableSubscription);
+
+ assertEquals(canGcExpected,
baseDestination.canGcConsumer.test(subscription));
+ }
+
+ static class MockConsumerInfo extends ConsumerInfo {
+
+ private final boolean networkSubscription;
+ private final boolean durableSubscription;
+
+ public MockConsumerInfo(ActiveMQDestination activemqDestination,
boolean networkSubscription, boolean durableSubscription) {
+ setDestination(activemqDestination);
+ this.networkSubscription = networkSubscription;
+ this.durableSubscription = durableSubscription;
+ }
+
+ @Override
+ public boolean isNetworkSubscription() {
+ return this.networkSubscription;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return this.durableSubscription;
+ }
+ }
+
+ static class MockSubscription extends QueueSubscription {
+
+ private final boolean wildcardSubscription;
+
+ public MockSubscription(ActiveMQDestination activemqDestination,
boolean networkSubscription, boolean wildcardSubscription, boolean
durableSubscription) throws Exception {
+ super(brokerService.getBroker(), null, null, new
MockConsumerInfo(activemqDestination, networkSubscription,
durableSubscription));
+ this.wildcardSubscription = wildcardSubscription;
+ }
+
+ @Override
+ public boolean isWildcard() {
+ return this.wildcardSubscription;
+ }
+ }
+}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCTest.java
index ee11bd27d9..061e68f330 100644
---
a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCTest.java
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCTest.java
@@ -18,13 +18,11 @@ package org.apache.activemq.broker.region;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
-import javax.jms.Connection;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
+import javax.jms.JMSException;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
@@ -34,19 +32,22 @@ import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.util.Wait;
-import org.apache.activemq.util.Wait.Condition;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
public class DestinationGCTest {
protected static final Logger logger =
LoggerFactory.getLogger(DestinationGCTest.class);
private final ActiveMQQueue queue = new ActiveMQQueue("TEST");
private final ActiveMQQueue otherQueue = new ActiveMQQueue("TEST-OTHER");
+ private final ActiveMQQueue wildcardQueueA = new
ActiveMQQueue("TEST.FOO.A");
+ private final ActiveMQQueue wildcardQueueB = new
ActiveMQQueue("TEST.FOO.B");
private BrokerService brokerService;
@@ -66,13 +67,30 @@ public class DestinationGCTest {
}
protected BrokerService createBroker() throws Exception {
- PolicyEntry entry = new PolicyEntry();
+ var entry = new PolicyEntry();
entry.setGcInactiveDestinations(true);
+ entry.setGcWithOnlyWildcardConsumers(true);
entry.setInactiveTimeoutBeforeGC(3000);
- PolicyMap map = new PolicyMap();
+ var map = new PolicyMap();
map.setDefaultEntry(entry);
- BrokerService broker = new BrokerService();
+ // GUARD.> queues allow wildcard-only removal but are excluded from the
+ // gc sweep, so the removeDestination guard tests are not raced by gc
+ var guardEntry = new PolicyEntry();
+ guardEntry.setQueue("GUARD.>");
+ guardEntry.setGcInactiveDestinations(false);
+ guardEntry.setGcWithOnlyWildcardConsumers(true);
+ map.put(new ActiveMQQueue("GUARD.>"), guardEntry);
+
+ // NOGC.> queues have the wildcard flag off - wildcard consumers keep
+ // these destinations active
+ var noGcEntry = new PolicyEntry();
+ noGcEntry.setQueue("NOGC.>");
+ noGcEntry.setGcInactiveDestinations(false);
+ noGcEntry.setGcWithOnlyWildcardConsumers(false);
+ map.put(new ActiveMQQueue("NOGC.>"), noGcEntry);
+
+ var broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(true);
broker.setDestinations(new ActiveMQDestination[] {queue});
@@ -87,39 +105,210 @@ public class DestinationGCTest {
public void testDestinationGCWithActiveConsumers() throws Exception {
assertEquals(1, brokerService.getAdminView().getQueues().length);
- ActiveMQConnectionFactory factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
- Connection connection = factory.createConnection();
- Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
- session.createProducer(otherQueue).close();
- MessageConsumer consumer = session.createConsumer(queue);
- consumer.setMessageListener(new MessageListener() {
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer = session.createConsumer(queue)) {
+
+ session.createProducer(otherQueue).close();
+ consumer.setMessageListener(message -> {});
+
+ connection.start();
+ assertTrue("After GC runs there should be one Queue.",
+ Wait.waitFor(() ->
brokerService.getAdminView().getQueues().length == 1));
+ }
+ }
+
+ @Test
+ public void testDestinationGCWithOnlyWildcardConsumers() throws Exception {
+ assertEquals(1, brokerService.getAdminView().getQueues().length);
+
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+
+ final var receivedCount = new AtomicInteger(0);
+
+ // Anonymous producer - does not register on the destinations, so the
gc
+ // assertions exercise only the wildcard consumer
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(null)) {
+
+ producer.send(wildcardQueueA, session.createTextMessage("Test
first step queueA"));
+ producer.send(wildcardQueueB, session.createTextMessage("Test
first step queueB"));
+
+ var consumer =
session.createConsumer(session.createQueue("TEST.FOO.*"));
+ consumer.setMessageListener(message ->
receivedCount.incrementAndGet());
+
+ connection.start();
+
+ // Confirm queues are gc'd
+ assertTrue("After GC runs there should be one Queue (count=" +
brokerService.getAdminView().getQueues().length + ")",
+ Wait.waitFor(() ->
brokerService.getAdminView().getQueues().length == 1, 30000, 1000));
- @Override
- public void onMessage(Message message) {
+ assertEquals(Integer.valueOf(2),
Integer.valueOf(receivedCount.get()));
+
+ // Confirm wild-card consumer is able to stay active after zero
matching destinations
+ producer.send(wildcardQueueA, session.createTextMessage("Test
second step queueA"));
+
+ // Confirm queues are gc'd
+ assertTrue("After GC runs there should be one Queue (count=" +
brokerService.getAdminView().getQueues().length + ")",
+ Wait.waitFor(() ->
brokerService.getAdminView().getQueues().length == 1, 30000, 1000));
+ assertEquals(Integer.valueOf(3),
Integer.valueOf(receivedCount.get()));
+ }
+ }
+
+ private int countMatchingTopics(String prefix) throws Exception {
+ var count = 0;
+ for (var name : brokerService.getAdminView().getTopics()) {
+ var destinationName = name.getKeyProperty("destinationName");
+ if (destinationName != null && destinationName.startsWith(prefix))
{
+ count++;
}
- });
+ }
+ return count;
+ }
+
+ // [AMQ-9692] non-durable topic flavor of the wildcard-consumer gc
lifecycle:
+ // topics are gc'd while the wildcard consumer stays connected, and
delivery
+ // continues when a producer recreates them
+ @Test(timeout = 60000)
+ public void testTopicDestinationGCWithOnlyWildcardConsumers() throws
Exception {
+ final var receivedCount = new AtomicInteger(0);
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+
+ // Subscribe before sending - non-durable topics do not retain
messages.
+ // Anonymous producer - does not register on the destinations
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer =
session.createConsumer(session.createTopic("TEST.BAR.*"));
+ var producer = session.createProducer(null)) {
+
+ consumer.setMessageListener(message ->
receivedCount.incrementAndGet());
+ connection.start();
+
+ producer.send(session.createTopic("TEST.BAR.A"),
session.createTextMessage("first-a"));
+ producer.send(session.createTopic("TEST.BAR.B"),
session.createTextMessage("first-b"));
+
+ assertTrue("Wildcard topic consumer should receive both messages",
+ Wait.waitFor(() -> receivedCount.get() == 2));
+
+ // Both topics should gc while the wildcard consumer stays
connected
+ assertTrue("After GC runs there should be no TEST.BAR. topics",
+ Wait.waitFor(() -> countMatchingTopics("TEST.BAR.") == 0,
30000, 1000));
- connection.start();
+ // A new send recreates the topic and the consumer must still
receive
+ producer.send(session.createTopic("TEST.BAR.A"),
session.createTextMessage("second-a"));
+ assertTrue("Wildcard topic consumer should receive after gc +
recreate",
+ Wait.waitFor(() -> receivedCount.get() == 3));
- assertTrue("After GC runs there should be one Queue.",
Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return brokerService.getAdminView().getQueues().length == 1;
+ assertTrue("Recreated topic should gc again",
+ Wait.waitFor(() -> countMatchingTopics("TEST.BAR.") == 0,
30000, 1000));
+ }
+ }
+
+ // [AMQ-9692] removeDestination(timeout=0) guard: an app consumer must
block removal
+ @Test(timeout = 60000)
+ public void testRemoveDestinationWithAppConsumerThrows() throws Exception {
+ var guardQueue = new ActiveMQQueue("GUARD.APP");
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer = session.createConsumer(guardQueue)) {
+
+ connection.start();
+
+ try {
+
brokerService.getBroker().removeDestination(brokerService.getAdminConnectionContext(),
guardQueue, 0);
+ fail("Expected JMSException removing a destination with an
active app consumer");
+ } catch (JMSException expected) {
+ assertTrue(expected.getMessage().contains("still has an active
subscription"));
}
- }));
+ }
+ }
+
+ // [AMQ-9692] removeDestination(timeout=0) guard: a wildcard-only consumer
permits
+ // removal when gcWithOnlyWildcardConsumers is enabled, and keeps working
afterwards
+ @Test(timeout = 60000)
+ public void testRemoveDestinationWithOnlyWildcardConsumerSucceeds() throws
Exception {
+ var guardQueue = new ActiveMQQueue("GUARD.WILD.A");
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ // Anonymous producer - does not register on the destination
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer =
session.createConsumer(session.createQueue("GUARD.WILD.*"));
+ var producer = session.createProducer(null)) {
+
+ connection.start();
+
+ producer.send(guardQueue,
session.createTextMessage("before-remove"));
+ assertNotNull("Wildcard consumer should receive",
consumer.receive(5000));
+
+ // Only the wildcard consumer remains - removal is allowed
+
brokerService.getBroker().removeDestination(brokerService.getAdminConnectionContext(),
guardQueue, 0);
+
+ // The wildcard consumer stays connected - a new send recreates
the queue
+ producer.send(guardQueue,
session.createTextMessage("after-remove"));
+ assertNotNull("Wildcard consumer should receive after remove +
recreate", consumer.receive(5000));
+ }
+ }
- connection.close();
+ // [AMQ-9692] removeDestination(timeout=0) guard: with
gcWithOnlyWildcardConsumers
+ // disabled a wildcard consumer blocks removal (default behavior unchanged)
+ @Test(timeout = 60000)
+ public void testRemoveDestinationWithWildcardConsumerFlagOffThrows()
throws Exception {
+ var noGcQueue = new ActiveMQQueue("NOGC.WILD.A");
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer =
session.createConsumer(session.createQueue("NOGC.WILD.*"));
+ var producer = session.createProducer(null)) {
+
+ connection.start();
+
+ producer.send(noGcQueue, session.createTextMessage("test"));
+ assertNotNull("Wildcard consumer should receive",
consumer.receive(5000));
+
+ try {
+
brokerService.getBroker().removeDestination(brokerService.getAdminConnectionContext(),
noGcQueue, 0);
+ fail("Expected JMSException removing a wildcard-consumed
destination with the flag off");
+ } catch (JMSException expected) {
+ assertTrue(expected.getMessage().contains("still has an active
subscription"));
+ }
+ }
+ }
+
+ // [AMQ-9692] removeDestination(timeout=0) guard: removing a nonexistent
destination
+ // is a silent no-op even when a wildcard subscription matches the name.
RegionBroker
+ // filters nonexistent destinations before delegating, so this contract is
only
+ // reachable at the region level - call the queue region directly.
+ @Test(timeout = 60000)
+ public void testRemoveNonExistentDestinationMatchingWildcardIsNoOp()
throws Exception {
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer =
session.createConsumer(session.createQueue("GUARD.MISSING.*"));
+ var producer = session.createProducer(null)) {
+
+ connection.start();
+
+ var queueCountBefore =
brokerService.getAdminView().getQueues().length;
+ var queueRegion = ((RegionBroker)
brokerService.getRegionBroker()).getQueueRegion();
+
queueRegion.removeDestination(brokerService.getAdminConnectionContext(),
+ new ActiveMQQueue("GUARD.MISSING.A"), 0);
+
+ assertEquals(queueCountBefore,
brokerService.getAdminView().getQueues().length);
+
+ // The wildcard subscription is unaffected by the no-op removal
+ producer.send(new ActiveMQQueue("GUARD.MISSING.A"),
session.createTextMessage("after-noop"));
+ assertNotNull("Wildcard consumer should still receive",
consumer.receive(5000));
+ }
}
@Test(timeout = 60000)
public void testDestinationGc() throws Exception {
assertEquals(1, brokerService.getAdminView().getQueues().length);
- assertTrue("After GC runs the Queue should be empty.",
Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return brokerService.getAdminView().getQueues().length == 0;
- }
- }));
+ assertTrue("After GC runs the Queue should be empty.",
+ Wait.waitFor(() -> brokerService.getAdminView().getQueues().length
== 0));
}
@Test(timeout = 60000)
@@ -131,48 +320,42 @@ public class DestinationGCTest {
brokerService.getAdminView().addQueue("TEST4");
assertEquals(5, brokerService.getAdminView().getQueues().length);
- Thread.sleep(7000);
-
- int queues = brokerService.getAdminView().getQueues().length;
- assertTrue(queues > 0 && queues < 5);
- assertTrue("After GC runs the Queue should be empty.",
Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return brokerService.getAdminView().getQueues().length == 0;
- }
- }));
+ // With maxPurgedDestinationsPerSweep=1, wait until at least one queue
has been GC'd
+ // but not all (verifying the sweep limit works)
+ assertTrue("GC should have removed some but not all queues",
+ Wait.waitFor(() -> {
+ final var count =
brokerService.getAdminView().getQueues().length;
+ return count > 0 && count < 5;
+ }, 15000, 500));
+
+ assertTrue("After GC runs the Queue should be empty.", Wait.waitFor(()
->
+ brokerService.getAdminView().getQueues().length == 0
+ , 30000, 500));
}
@Test(timeout = 60000)
public void testDestinationGcAnonymousProducer() throws Exception {
- final ActiveMQQueue q = new ActiveMQQueue("Q.TEST.ANONYMOUS.PRODUCER");
+ final var q = new ActiveMQQueue("Q.TEST.ANONYMOUS.PRODUCER");
brokerService.getAdminView().addQueue(q.getPhysicalName());
assertEquals(2, brokerService.getAdminView().getQueues().length);
- final ActiveMQConnectionFactory factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
- final Connection connection = factory.createConnection();
- final Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ final var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ try (var connection = factory.createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(null)) {
- // wait for the queue to be marked for GC
- logger.info("Waiting for '{}' to be marked for GC...", q);
- Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return brokerService.getDestination(q).canGC();
- }
- }, Wait.MAX_WAIT_MILLIS, 500L);
-
- // create anonymous producer and send a message
- logger.info("Sending PERSISTENT message to QUEUE '{}'",
q.getPhysicalName());
- final MessageProducer producer = session.createProducer(null);
- producer.send(q, session.createTextMessage());
- producer.close();
+ // wait for the queue to be marked for GC
+ logger.info("Waiting for '{}' to be marked for GC...", q);
+ Wait.waitFor(() -> brokerService.getDestination(q).canGC(),
Wait.MAX_WAIT_MILLIS, 500L);
- assertFalse(brokerService.getDestination(q).canGC());
+ // send a message via the anonymous producer
+ logger.info("Sending PERSISTENT message to QUEUE '{}'",
q.getPhysicalName());
+ producer.send(q, session.createTextMessage());
- connection.close();
+ assertFalse(brokerService.getDestination(q).canGC());
+ }
}
}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCWildcardDurableSubTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCWildcardDurableSubTest.java
new file mode 100644
index 0000000000..c8c876931d
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationGCWildcardDurableSubTest.java
@@ -0,0 +1,360 @@
+/**
+ * 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.broker.region;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import javax.jms.Connection;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.store.TopicMessageStore;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * [AMQ-9692] Verify destination GC behavior for topics whose only
+ * subscription is a durable *wildcard* subscriber.
+ *
+ * Durable subscription state (registration + pending messages) is kept in the
+ * per-topic message store, and TopicRegion.addSubscriptionsForDestination()
+ * recovers durable subs from that store when a destination is (re)created.
+ * Destination GC destroys the store, so these tests probe whether the
+ * durable-subscription guarantees survive the gc + recreate cycle.
+ */
+public class DestinationGCWildcardDurableSubTest {
+
+ private static final String CLIENT_ID = "durable-wildcard-client";
+ private static final String SUB_NAME = "durable-wildcard-sub";
+ private static final ActiveMQTopic WILDCARD_TOPIC = new
ActiveMQTopic("TEST.DUR.>");
+ private static final ActiveMQTopic TOPIC_A = new
ActiveMQTopic("TEST.DUR.A");
+
+ private BrokerService brokerService;
+
+ @Before
+ public void setUp() throws Exception {
+ brokerService = createBroker();
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (brokerService != null) {
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ }
+ }
+
+ protected BrokerService createBroker() throws Exception {
+ return createBroker(true);
+ }
+
+ protected BrokerService createBroker(boolean deleteAllMessagesOnStartup)
throws Exception {
+ return createBroker(deleteAllMessagesOnStartup, true);
+ }
+
+ protected BrokerService createBroker(boolean deleteAllMessagesOnStartup,
boolean gcWithOnlyWildcardConsumers) throws Exception {
+ var entry = new PolicyEntry();
+ entry.setGcInactiveDestinations(true);
+ entry.setGcWithOnlyWildcardConsumers(gcWithOnlyWildcardConsumers);
+ entry.setInactiveTimeoutBeforeGC(1000);
+ var map = new PolicyMap();
+ map.setDefaultEntry(entry);
+
+ var broker = new BrokerService();
+ // Persistent so durable subscription state goes through a real store
+ broker.setPersistent(true);
+
broker.setDataDirectory("target/activemq-data/DestinationGCWildcardDurableSubTest");
+ broker.setDeleteAllMessagesOnStartup(deleteAllMessagesOnStartup);
+ broker.setUseJmx(true);
+ broker.setSchedulePeriodForDestinationPurge(500);
+ broker.setDestinationPolicy(map);
+ return broker;
+ }
+
+ private Connection createConnection() throws Exception {
+ var factory = new
ActiveMQConnectionFactory("vm://localhost?create=false");
+ var connection = factory.createConnection();
+ connection.setClientID(CLIENT_ID);
+ connection.start();
+ return connection;
+ }
+
+ private int countMatchingTopics() throws Exception {
+ var count = 0;
+ for (var name : brokerService.getAdminView().getTopics()) {
+ var destinationName = name.getKeyProperty("destinationName");
+ if (destinationName != null &&
destinationName.startsWith("TEST.DUR.")) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ /**
+ * Asserts the topic survives several gc sweep periods. The sweep runs
every
+ * 500ms with a 1000ms inactive timeout, so 4s covers multiple full
+ * mark-and-collect cycles.
+ */
+ private void assertTopicNotGcd() throws Exception {
+ Thread.sleep(4000);
+ assertEquals("Topic with a durable subscription must not be gc'd", 1,
countMatchingTopics());
+ }
+
+ /**
+ * A durable subscription's registration and pending messages live in the
+ * topic's store, which gc destroys. An active durable wildcard subscriber
+ * must therefore prevent the matched topic from being gc'd, and delivery
+ * must continue working.
+ */
+ @Test(timeout = 60000)
+ public void testActiveDurableWildcardSubPreventsTopicGc() throws Exception
{
+ // Anonymous producer - does not register on the destination, so the gc
+ // assertions exercise only the durable consumer
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME);
+ var producer = session.createProducer(null)) {
+
+ producer.send(TOPIC_A, session.createTextMessage("first"));
+
+ var received = durableSubscriber.receive(5000);
+ assertNotNull("Durable wildcard sub should receive first message",
received);
+ assertEquals("first", ((TextMessage) received).getText());
+
+ // Topic is drained but has a durable wildcard consumer - must not
gc
+ assertTopicNotGcd();
+
+ producer.send(TOPIC_A, session.createTextMessage("second"));
+
+ received = durableSubscriber.receive(5000);
+ assertNotNull("Durable wildcard sub should continue receiving
messages", received);
+ assertEquals("second", ((TextMessage) received).getText());
+
+ durableSubscriber.close();
+ session.unsubscribe(SUB_NAME);
+ }
+ }
+
+ /**
+ * A topic holding a message pending for an offline durable wildcard
+ * subscriber must NOT be gc'd - the pending message keeps the destination
+ * message count non-zero.
+ */
+ @Test(timeout = 60000)
+ public void testOfflineDurableWildcardSubWithPendingMessageNotGcd() throws
Exception {
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(TOPIC_A)) {
+
+ session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME).close();
+
+ // Subscriber offline (closed); send a message it should receive
later
+ producer.send(session.createTextMessage("pending"));
+ }
+
+ // Give the gc sweep several periods to (incorrectly) collect the topic
+ assertTopicNotGcd();
+
+ // Reconnect and confirm the pending message is delivered
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME)) {
+
+ var received = durableSubscriber.receive(5000);
+ assertNotNull("Offline durable sub should receive pending message
on reconnect", received);
+ assertEquals("pending", ((TextMessage) received).getText());
+
+ durableSubscriber.close();
+ session.unsubscribe(SUB_NAME);
+ }
+ }
+
+ /**
+ * Durability across broker restart: the durable wildcard sub's
registration is
+ * recovered from the topic's store at startup as an INACTIVE subscription
-
+ * counted on the destination, but not present in its consumers list. The
+ * recovered topic must not be gc'd, and a message sent before the
subscriber
+ * reconnects must be delivered.
+ */
+ @Test(timeout = 60000)
+ public void testOfflineDurableWildcardSubSurvivesBrokerRestart() throws
Exception {
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME);
+ var producer = session.createProducer(TOPIC_A)) {
+
+ producer.send(session.createTextMessage("before-restart"));
+ assertNotNull(durableSubscriber.receive(5000));
+ }
+
+ // Restart without wiping the store - the topic is recovered with an
+ // inactive durable subscription
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ brokerService = createBroker(false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+
+ // The recovered topic holds the durable registration - must not gc
+ assertTopicNotGcd();
+
+ // Send while the durable sub is still offline
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(TOPIC_A)) {
+
producer.send(session.createTextMessage("while-offline-after-restart"));
+ }
+
+ // Reconnect - durability requires the message survive the restart +
gc sweeps
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME)) {
+
+ var received = durableSubscriber.receive(5000);
+ assertNotNull("Message sent while durable sub was offline across a
restart must be delivered", received);
+ assertEquals("while-offline-after-restart", ((TextMessage)
received).getText());
+
+ durableSubscriber.close();
+ session.unsubscribe(SUB_NAME);
+ }
+ }
+
+ /**
+ * Pre-change behavior: with gcWithOnlyWildcardConsumers disabled (the
default),
+ * a durable wildcard subscriber keeps its matched topics from being gc'd -
+ * active or offline - and durability holds. Once the subscription is
removed
+ * entirely, gcInactiveDestinations collects the abandoned topic as before.
+ * Also documents the registration model: the durable sub is persisted in
each
+ * concrete matching topic's store, recorded under its wildcard
destination -
+ * no destination or store exists for the wildcard name itself.
+ */
+ @Test(timeout = 60000)
+ public void testDurableWildcardSubWithGcWildcardDisabled() throws
Exception {
+ // Swap in a broker with the wildcard gc flag OFF (pre-change /
default config)
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ brokerService = createBroker(true, false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+
+ // Anonymous producer - does not register on the destination, so the gc
+ // assertions exercise only the durable consumer
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME);
+ var producer = session.createProducer(null)) {
+
+ producer.send(TOPIC_A, session.createTextMessage("first"));
+ assertNotNull(durableSubscriber.receive(5000));
+
+ // The durable sub is registered in the concrete topic's store
under its
+ // wildcard subscribed destination
+ var store = (TopicMessageStore)
brokerService.getDestination(TOPIC_A).getMessageStore();
+ var subscriptions = store.getAllSubscriptions();
+ assertEquals(1, subscriptions.length);
+ assertEquals(WILDCARD_TOPIC,
subscriptions[0].getSubscribedDestination());
+
+ // Active durable wildcard sub - topic must not gc with the flag
off
+ assertTopicNotGcd();
+ }
+
+ // Offline durable wildcard sub - topic must still not gc
+ assertTopicNotGcd();
+
+ // Send while the durable sub is offline
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(TOPIC_A)) {
+ producer.send(session.createTextMessage("while-offline"));
+ }
+
+ // Reconnect - durability held
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME)) {
+
+ var received = durableSubscriber.receive(5000);
+ assertNotNull("Message sent while durable sub was offline must be
delivered", received);
+ assertEquals("while-offline", ((TextMessage) received).getText());
+
+ // Remove the subscription entirely - the topic is now truly
abandoned
+ durableSubscriber.close();
+ session.unsubscribe(SUB_NAME);
+ }
+
+ // With no durable registration left, gcInactiveDestinations collects
the topic
+ assertTrue("Abandoned topic should be gc'd once the durable
subscription is removed",
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return countMatchingTopics() == 0;
+ }
+ }, 15000, 500));
+ }
+
+ /**
+ * Durability guarantee: durable wildcard sub goes offline with the topic
+ * fully drained. The empty topic must NOT be gc'd (its store holds the
+ * durable registration), and a message sent while the subscriber is
offline
+ * MUST be delivered when it reconnects.
+ */
+ @Test(timeout = 60000)
+ public void
testOfflineDurableWildcardSubEmptyTopicNotGcdAndDeliversOnReconnect() throws
Exception {
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME);
+ var producer = session.createProducer(TOPIC_A)) {
+
+ producer.send(session.createTextMessage("before-offline"));
+ assertNotNull(durableSubscriber.receive(5000));
+ }
+
+ // Topic is drained and the durable wildcard sub is offline - must not
gc
+ assertTopicNotGcd();
+
+ // Send a message while the durable sub is offline
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(TOPIC_A)) {
+ producer.send(session.createTextMessage("while-offline"));
+ }
+
+ // Reconnect the durable subscriber - durability requires the message
arrive
+ try (var connection = createConnection();
+ var session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var durableSubscriber =
session.createDurableSubscriber(WILDCARD_TOPIC, SUB_NAME)) {
+
+ var received = durableSubscriber.receive(5000);
+ assertNotNull("Message sent while durable sub was offline must be
delivered on reconnect", received);
+ assertEquals("while-offline", ((TextMessage) received).getText());
+
+ durableSubscriber.close();
+ session.unsubscribe(SUB_NAME);
+ }
+ }
+}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationIsActiveTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationIsActiveTest.java
new file mode 100644
index 0000000000..7e430b3bd5
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/DestinationIsActiveTest.java
@@ -0,0 +1,341 @@
+/**
+ * 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.broker.region;
+
+import static org.junit.Assert.*;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.ConnectionId;
+import org.apache.activemq.command.ConsumerId;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.command.ProducerInfo;
+import org.apache.activemq.command.SessionId;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * [AMQ-9692] Support garbage collecting destinations
+ * that have a wildcard-only subscription.
+ *
+ * This test suite confirms the logic in the
+ * BaseDestination.isActive() method to ensure
+ * destinations are not accidentally deleted due
+ * to incorrect logic combination of PolicyEntr
+ * config flag and status of current subscriptions.
+ *
+ * Every row runs against both a queue and a (non-durable) topic destination -
+ * the isActive() contract is destination-type agnostic. Note the durable rows
+ * exercise different paths per type: on a queue the durable sub appears in the
+ * consumers list (predicate path); on a topic an inactive durable sub is
+ * counted but not listed (count/list mismatch path).
+ *
+ * prod = attached producer
+ * appC = normal application consumer
+ * netC = network consumer
+ * wildC = wildcard consumer (non-durable)
+ * durWildC = durable wildcard consumer
+ */
+@RunWith(Parameterized.class)
+public class DestinationIsActiveTest {
+
+ private static BrokerService brokerService;
+ private static final AtomicInteger counter = new AtomicInteger(0);
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ brokerService = createBroker();
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ if (brokerService != null) {
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+ }
+ }
+
+ @Parameterized.Parameters(name = "dest={0} gcNC={1} gcWC={2} prod={3}
appC={4} netC={5} wildC={6} durWildC={7} exp={8}") // Optional name attribute
for better test reporting
+ public static Collection<Object[]> data() {
+ // The truth table is destination-type agnostic - run every row against
+ // both a queue and a topic destination
+ var truthTable = new Object[][] {
+ // Simple app consumer
+ { false, false, false, false, false, false, false, false },
+ { false, true, false, false, false, false, false, false },
+ { true, false, false, false, false, false, false, false },
+ { true, true, false, false, false, false, false, false },
+ { false, false, false, true, false, false, false, true },
+ { false, true, false, true, false, false, false, true },
+ { true, false, false, true, false, false, false, true },
+ { true, true, false, true, false, false, false, true },
+
+ // Network consumer
+ { false, false, false, false, true, false, false, true },
+ { false, true, false, false, true, false, false, true },
+ { true, false, false, false, true, false, false, false },
+ { true, true, false, false, true, false, false, false },
+ { false, false, false, true, true, false, false, true },
+ { false, true, false, true, true, false, false, true },
+ { true, false, false, true, true, false, false, true },
+ { true, true, false, true, true, false, false, true },
+
+ // Wildcard consumer
+ { false, false, false, false, false, true, false, true },
+ { false, true, false, false, false, true, false, false },
+ { true, false, false, false, false, true, false, true },
+ { true, true, false, false, false, true, false, false },
+ { false, false, false, true, false, true, false, true },
+ { false, true, false, true, false, true, false, true },
+ { true, false, false, true, false, true, false, true },
+ { true, true, false, true, false, true, false, true },
+
+ // Mixed network + wildcard consumers - gc only allowed when
+ // BOTH flags permit ignoring their respective consumer type
+ { false, false, false, false, true, true, false, true },
+ { false, true, false, false, true, true, false, true },
+ { true, false, false, false, true, true, false, true },
+ { true, true, false, false, true, true, false, false },
+ { false, false, false, true, true, true, false, true },
+ { false, true, false, true, true, true, false, true },
+ { true, false, false, true, true, true, false, true },
+ { true, true, false, true, true, true, false, true },
+
+ // Durable wildcard consumer - never gc-eligible, its
registration
+ // and pending messages live in the destination's store
+ { false, false, false, false, false, false, true, true },
+ { false, true, false, false, false, false, true, true },
+ { true, false, false, false, false, false, true, true },
+ { true, true, false, false, false, false, true, true },
+ { false, true, false, false, false, true, true, true },
+ { true, true, false, false, true, true, true, true },
+
+ // Attached producer - always active, even when every consumer
+ // present is gc-eligible under the enabled flags
+ { false, false, true, false, false, false, false, true },
+ { true, true, true, false, false, false, false, true },
+ { true, false, true, false, true, false, false, true },
+ { false, true, true, false, false, true, false, true },
+ { true, true, true, false, true, true, false, true }
+ };
+
+ var params = new ArrayList<Object[]>();
+ for (var destinationType : List.of("queue", "topic")) {
+ for (var row : truthTable) {
+ var param = new Object[row.length + 1];
+ param[0] = destinationType;
+ System.arraycopy(row, 0, param, 1, row.length);
+ params.add(param);
+ }
+ }
+ return params;
+ }
+
+ private final String destinationType;
+ private final boolean gcWithNetworkConsumersEnabled;
+ private final boolean gcWithOnlyWildcardConsumersEnabled;
+ private final boolean producerActive;
+ private final boolean appConsumerActive;
+ private final boolean networkConsumerActive;
+ private final boolean wildcardConsumerActive;
+ private final boolean durableWildcardConsumerActive;
+ private final boolean activeExpected;
+
+ public DestinationIsActiveTest(String destinationType, boolean
gcWithNetworkConsumersEnabled, boolean gcWithOnlyWildcardConsumersEnabled,
boolean producerActive, boolean appConsumerActive, boolean
networkConsumerActive, boolean wildcardConsumerActive, boolean
durableWildcardConsumerActive, boolean activeExpected) {
+ this.destinationType = destinationType;
+ this.gcWithNetworkConsumersEnabled = gcWithNetworkConsumersEnabled;
+ this.gcWithOnlyWildcardConsumersEnabled =
gcWithOnlyWildcardConsumersEnabled;
+ this.producerActive = producerActive;
+ this.appConsumerActive = appConsumerActive;
+ this.networkConsumerActive = networkConsumerActive;
+ this.wildcardConsumerActive = wildcardConsumerActive;
+ this.durableWildcardConsumerActive = durableWildcardConsumerActive;
+ this.activeExpected = activeExpected;
+ }
+
+ @Test
+ public void testDestinationIsActive() throws Exception {
+ var destinationName = "amq.gc." + counter.incrementAndGet();
+ final var isTopic = "topic".equals(destinationType);
+
+ var policyEntry = new PolicyEntry();
+ policyEntry.setGcInactiveDestinations(true);
+
policyEntry.setGcWithOnlyWildcardConsumers(gcWithOnlyWildcardConsumersEnabled);
+ policyEntry.setGcWithNetworkConsumers(gcWithNetworkConsumersEnabled);
+ policyEntry.setInactiveTimeoutBeforeGC(3000L);
+ if (isTopic) {
+ policyEntry.setTopic(destinationName);
+ } else {
+ policyEntry.setQueue(destinationName);
+ }
+
brokerService.getDestinationPolicy().setPolicyEntries(List.of(policyEntry));
+
+ ActiveMQDestination activemqDestination;
+ if (isTopic) {
+ brokerService.getAdminView().addTopic(destinationName);
+ activemqDestination = new ActiveMQTopic(destinationName);
+ } else {
+ brokerService.getAdminView().addQueue(destinationName);
+ activemqDestination = new ActiveMQQueue(destinationName);
+ }
+ var destination = brokerService.getDestination(activemqDestination);
+
+ assertFalse(destination.isActive());
+
+ if(producerActive) {
+ destination.addProducer(null, new ProducerInfo());
+ }
+ if(appConsumerActive) {
+ destination.addSubscription(null, new
MockQueueSubscription(activemqDestination, false, false, false));
+ }
+ if(networkConsumerActive) {
+ destination.addSubscription(null, new
MockQueueSubscription(activemqDestination, true, false, false));
+ }
+ if(wildcardConsumerActive) {
+ destination.addSubscription(null, new
MockQueueSubscription(activemqDestination, false, true, false));
+ }
+ if(durableWildcardConsumerActive) {
+ // Topic.addSubscription casts durable subscriptions, so topics
need
+ // a DurableTopicSubscription-based mock (inactive, like one
recovered
+ // from the store at broker start)
+ if (isTopic) {
+ destination.addSubscription(null, new
MockDurableTopicSubscription(activemqDestination, true));
+ } else {
+ destination.addSubscription(null, new
MockQueueSubscription(activemqDestination, false, true, true));
+ }
+ }
+
+ assertEquals(activeExpected, destination.isActive());
+
+ // Test parameter config safety checks
+ // if an appConsumer is active, destination must *always* be active
+ if(appConsumerActive) {
+ assertTrue(destination.isActive());
+ }
+ // if a producer is attached, destination must *always* be active
+ if(producerActive) {
+ assertTrue(destination.isActive());
+ }
+ // a durable subscription must *always* keep the destination active
+ if(durableWildcardConsumerActive) {
+ assertTrue(destination.isActive());
+ }
+
+ if (isTopic) {
+ brokerService.getAdminView().removeTopic(destinationName);
+ } else {
+ brokerService.getAdminView().removeQueue(destinationName);
+ }
+ }
+
+ protected static BrokerService createBroker() throws Exception {
+ var map = new PolicyMap();
+ map.setDefaultEntry(new PolicyEntry());
+
+ var broker = new BrokerService();
+ broker.setPersistent(false);
+ broker.setUseJmx(true);
+ broker.setSchedulePeriodForDestinationPurge(100_000_000);
+ broker.setSchedulerSupport(true);
+ broker.setMaxPurgedDestinationsPerSweep(1);
+ broker.setDestinationPolicy(map);
+ return broker;
+ }
+
+ static class MockConsumerInfo extends ConsumerInfo {
+
+ private final boolean networkSubscription;
+ private final boolean durableSubscription;
+
+ public MockConsumerInfo(ActiveMQDestination activeMQDestination,
boolean networkSubscription, boolean durableSubscription) {
+ setDestination(activeMQDestination);
+ this.networkSubscription = networkSubscription;
+ this.durableSubscription = durableSubscription;
+ }
+
+ @Override
+ public boolean isNetworkSubscription() {
+ return this.networkSubscription;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return this.durableSubscription;
+ }
+ }
+
+ static class MockQueueSubscription extends QueueSubscription {
+
+ private final boolean wildCardSubscription;
+
+ public MockQueueSubscription(ActiveMQDestination activemqDestination,
boolean networkSubscription, boolean wildCardSubscription, boolean
durableSubscription) throws Exception {
+ super(brokerService.getBroker(), null, null, new
MockConsumerInfo(activemqDestination, networkSubscription,
durableSubscription));
+ this.wildCardSubscription = wildCardSubscription;
+ }
+
+ @Override
+ public boolean isWildcard() {
+ return this.wildCardSubscription;
+ }
+ }
+
+ // An inactive durable subscription, as recovered from the store at broker
start
+ static class MockDurableTopicSubscription extends DurableTopicSubscription
{
+
+ private final boolean wildCardSubscription;
+
+ public MockDurableTopicSubscription(ActiveMQDestination
activemqDestination, boolean wildCardSubscription) throws Exception {
+ super(brokerService.getBroker(), brokerService.getSystemUsage(),
+ durableContext(),
durableConsumerInfo(activemqDestination), true);
+ this.wildCardSubscription = wildCardSubscription;
+ }
+
+ @Override
+ public boolean isWildcard() {
+ return this.wildCardSubscription;
+ }
+
+ private static ConnectionContext durableContext() throws Exception {
+ var context = new ConnectionContext();
+ context.setClientId("mock-durable-client-" + counter.get());
+ context.setBroker(brokerService.getBroker());
+ return context;
+ }
+
+ private static ConsumerInfo durableConsumerInfo(ActiveMQDestination
activemqDestination) {
+ var info = new MockConsumerInfo(activemqDestination, false, true);
+ info.setSubscriptionName("mock-durable-sub");
+ info.setConsumerId(new ConsumerId(new SessionId(new
ConnectionId("mock-durable-connection"), 1), counter.get()));
+ return info;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
