cshannon commented on code in PR #1484:
URL: https://github.com/apache/activemq/pull/1484#discussion_r3904489432
##########
activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java:
##########
@@ -260,16 +260,21 @@ protected List<Subscription>
addSubscriptionsForDestination(ConnectionContext co
}
@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) {
+ final Destination dest = destinations.get(destination);
+ final boolean destActive = dest != null && dest.isActive();
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);
+ if (sub.matches(destination)) {
+ if (dest == null) {
Review Comment:
This null check doesn't make sense, sub.matches(null) will throw an NPE and
I would get rid of it entirely. There could be race conditions with destination
removal where it doesn't exist in the map anymore because you are not under
lock. If you look further down the code will just only process if not null.
I would simply this entire block of code and do the following:
```java
if (timeout == 0) {
final Destination dest = destinations.get(destination);
if (dest != null) {
for (Subscription sub : subscriptions.values()) {
if (sub.matches(destination) && dest.isActive()) {
throw new JMSException("Destination: " + destination + "
still has an active subscription: " + sub);
}
}
}
}
```
##########
activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java:
##########
@@ -311,12 +313,42 @@ public final MessageStore getMessageStore() {
@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;
+ }
+
+ protected Predicate<Subscription> canGcConsumer = subscription -> {
Review Comment:
this should likely be private and final as there is no need to override it
for now.
##########
activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java:
##########
@@ -311,12 +313,42 @@ public final MessageStore getMessageStore() {
@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;
+ }
+
+ protected 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());
Review Comment:
I thought you were getting rid of this durable check here as it isn't
necessary if relying on `keepDurableSubsActive` flag?
##########
activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java:
##########
@@ -796,19 +828,37 @@ public boolean isGcWithNetworkConsumers() {
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
+ && !isActive()) {
Review Comment:
I don't think this change is needed, you don't need to change canGC() at all
or add the isActive() check.
The way this is supposed to work is `markForGC() `does all that work and
then sets `lastActiveTime`, so this method only needs to check that. I would
not change canGC() method at all
##########
activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java:
##########
@@ -936,6 +936,13 @@ protected void purgeInactiveDestinations() {
context.setBroker(this);
for (Destination dest : list) {
+ // Re-verify eligibility right before removal - a message
can arrive
+ // (e.g. from an anonymous producer, which does not hold
the purge
+ // lock) between the mark loop above and this removal loop.
+ if (!dest.canGC()) {
Review Comment:
This is also unnecessary and should be removed, it breaks the
`maxPurgedDests` check as we may now not purge the max that we want.
When this loop executes it's executing immediately after the loop runs to
check and mark for GC and also count the number of destinations for
maxPurgeDests. No matter what there is a race condition here, this isn't going
to fix that and it doesn't make sense to check like 2 milliseconds later the
same condition.
--
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
