AMQ-6901 - Make sure proper policy is used to configure a destination

When multiple wildcard policies exist in a hierarchy it was possible for
the wrong policy to be selected when configuring a destination

(cherry picked from commit 2ce1c1352f)
This commit is contained in:
Christopher L. Shannon (cshannon) 2018-02-16 09:35:16 -05:00
parent 4841ff56e3
commit 3c2d5a3015
2 changed files with 72 additions and 6 deletions

View File

@ -23,6 +23,8 @@ import java.util.List;
import java.util.Set; import java.util.Set;
import java.util.SortedSet; import java.util.SortedSet;
import java.util.TreeSet; import java.util.TreeSet;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.command.ActiveMQDestination;
@ -46,6 +48,7 @@ public class DestinationMap {
private DestinationMapNode topicRootNode = new DestinationMapNode(null); private DestinationMapNode topicRootNode = new DestinationMapNode(null);
private DestinationMapNode tempTopicRootNode = new DestinationMapNode(null); private DestinationMapNode tempTopicRootNode = new DestinationMapNode(null);
/** /**
* Looks up the value(s) matching the given Destination key. For simple * Looks up the value(s) matching the given Destination key. For simple
* destinations this is typically a List of one single value, for wildcards * destinations this is typically a List of one single value, for wildcards
@ -202,19 +205,34 @@ public class DestinationMap {
* @return the largest matching value or null if no value matches * @return the largest matching value or null if no value matches
*/ */
@SuppressWarnings({"rawtypes", "unchecked"}) @SuppressWarnings({"rawtypes", "unchecked"})
public Object chooseValue(final ActiveMQDestination destination) { public DestinationMapEntry chooseValue(final ActiveMQDestination destination) {
Set set = get(destination); Set<DestinationMapEntry> set = get(destination);
if (set == null || set.isEmpty()) { if (set == null || set.isEmpty()) {
return null; return null;
} }
SortedSet sortedSet = new TreeSet(new Comparator<DestinationMapEntry>() {
//Comparator to sort in order - we want to pick the exact match by destination or the
//closest parent that applies
final Comparator<DestinationMapEntry> comparator = new Comparator<DestinationMapEntry>() {
@Override @Override
public int compare(DestinationMapEntry entry1, DestinationMapEntry entry2) { public int compare(DestinationMapEntry entry1, DestinationMapEntry entry2) {
return destination.equals(entry1.destination) ? -1 : (destination.equals(entry2.destination) ? 1 : entry1.compareTo(entry2)); return destination.equals(entry1.destination) ? -1 : (destination.equals(entry2.destination) ? 1 : entry1.compareTo(entry2));
} }
}); };
sortedSet.addAll(set);
return sortedSet.first(); //Sort and filter out any children and non matching entries
final SortedSet<DestinationMapEntry> sortedSet = set.stream()
.filter(entry -> isMatchOrParent(destination, (DestinationMapEntry)entry))
.collect(Collectors.toCollection(() -> new TreeSet<DestinationMapEntry>(comparator)));
return sortedSet.size() > 0 ? sortedSet.first() : null;
}
@SuppressWarnings("rawtypes")
//Used to filter out any child/unmatching entries
private boolean isMatchOrParent(final ActiveMQDestination destination, final DestinationMapEntry entry) {
final DestinationFilter filter = DestinationFilter.parseFilter(entry.getDestination());
return destination.equals(entry.getDestination()) || filter.matches(destination);
} }
/** /**

View File

@ -457,6 +457,54 @@ public class JavaPolicyEntryTest extends RuntimeConfigTestSupport {
verifyQueueLimit("queue.test", 1024); verifyQueueLimit("queue.test", 1024);
} }
@Test
public void testModWithChildWildcardPolicies() throws Exception {
BrokerService brokerService = new BrokerService();
PolicyMap policyMap = new PolicyMap();
PolicyEntry entry = new PolicyEntry();
entry.setQueue(">");
entry.setMemoryLimit(1024);
PolicyEntry entry2 = new PolicyEntry();
entry2.setQueue("queue.child.>");
entry2.setMemoryLimit(2048);
PolicyEntry entry3 = new PolicyEntry();
entry3.setQueue("queue.child.one.>");
entry3.setMemoryLimit(4096);
policyMap.setPolicyEntries(Arrays.asList(entry, entry2, entry3));
brokerService.setDestinationPolicy(policyMap);
startBroker(brokerService);
assertTrue("broker alive", brokerService.isStarted());
brokerService.getBroker().addDestination(
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.>"), false);
brokerService.getBroker().addDestination(
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.>"), false);
brokerService.getBroker().addDestination(
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.one.>"), false);
brokerService.getBroker().addDestination(
brokerService.getAdminConnectionContext(), new ActiveMQQueue("queue.child.one"), false);
//check destinations before policy updates
verifyQueueLimit("queue.>", 1024);
verifyQueueLimit("queue.child.>", 2048);
verifyQueueLimit("queue.child.one", 4096);
//Reapply new limit to policy 2
entry2.setMemoryLimit(4194304);
javaConfigBroker.modifyPolicyEntry(entry2);
TimeUnit.SECONDS.sleep(SLEEP);
//verify that destination at a higher level policy is not affected
verifyQueueLimit("queue.>", 1024);
verifyQueueLimit("queue.child.>", 4194304);
verifyQueueLimit("queue.child.one.>", 4096);
verifyQueueLimit("queue.child.one", 4096);
}
@Test @Test
public void testModParentPolicy() throws Exception { public void testModParentPolicy() throws Exception {
BrokerService brokerService = new BrokerService(); BrokerService brokerService = new BrokerService();