https://issues.apache.org/jira/browse/AMQ-5938 - remove unnecessary topic cast - breaks auth scenario. Fix and test

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java

@@ -565,7 +565,7 @@ public class ManagedRegionBroker extends RegionBroker {
     public void remove(SubscriptionView view, String messageId)  throws Exception {
         ActiveMQDestination destination = getTopicDestination(view);
         if (destination != null) {
-            final Topic topic = (Topic) getTopicRegion().getDestinationMap().get(destination);
+            final Destination topic = getTopicRegion().getDestinationMap().get(destination);
             final MessageAck messageAck = new MessageAck();
             messageAck.setMessageID(new MessageId(messageId));
             messageAck.setDestination(destination);
@@ -588,7 +588,7 @@ public class ManagedRegionBroker extends RegionBroker {
     protected Message[] getSubscriberMessages(SubscriptionView view) {
         ActiveMQDestination destination = getTopicDestination(view);
         if (destination != null) {
-            Topic topic = (Topic) getTopicRegion().getDestinationMap().get(destination);
+            Destination topic = getTopicRegion().getDestinationMap().get(destination);
             return topic.browse();
 
         } else {

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineBrowseRemoveTest.java

@@ -29,8 +29,15 @@ import javax.management.ObjectName;
 import javax.management.openmbean.CompositeData;
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.TestSupport;
+import org.apache.activemq.broker.BrokerPlugin;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.jmx.DurableSubscriptionViewMBean;
+import org.apache.activemq.filter.DestinationMapEntry;
+import org.apache.activemq.security.AuthenticationUser;
+import org.apache.activemq.security.AuthorizationEntry;
+import org.apache.activemq.security.AuthorizationPlugin;
+import org.apache.activemq.security.DefaultAuthorizationMap;
+import org.apache.activemq.security.SimpleAuthenticationPlugin;
 import org.apache.activemq.store.PersistenceAdapter;
 import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
 import org.junit.Test;
@@ -48,6 +55,7 @@ public class DurableSubscriptionOfflineBrowseRemoveTest extends DurableSubscript
 
     private static final Logger LOG = LoggerFactory.getLogger(DurableSubscriptionOfflineBrowseRemoveTest.class);
 
+    public static final String IDENTITY = "milly";
     public boolean keepDurableSubsActive;
 
     @Parameterized.Parameters(name = "PA-{0}.KeepSubsActive-{1}")
@@ -69,6 +77,29 @@ public class DurableSubscriptionOfflineBrowseRemoveTest extends DurableSubscript
         this.keepDurableSubsActive = keepDurableSubsActive;
     }
 
+    @Override
+    public void configurePlugins(BrokerService brokerService) throws Exception {
+        List<DestinationMapEntry> authorizationEntries = new ArrayList<>();
+
+        AuthorizationEntry entry = new AuthorizationEntry();
+        entry.setTopic(">");
+        entry.setRead(IDENTITY);
+        entry.setWrite(IDENTITY);
+        entry.setAdmin(IDENTITY);
+        authorizationEntries.add(entry);
+
+        DefaultAuthorizationMap authorizationMap = new DefaultAuthorizationMap(authorizationEntries);
+        AuthorizationPlugin authorizationPlugin = new AuthorizationPlugin(authorizationMap);
+
+        List<AuthenticationUser> users = new ArrayList<>();
+        users.add(new AuthenticationUser(IDENTITY, IDENTITY, IDENTITY));
+
+        SimpleAuthenticationPlugin authenticationPlugin = new SimpleAuthenticationPlugin(users);
+
+
+        broker.setPlugins(new BrokerPlugin[]{authenticationPlugin, authorizationPlugin});
+
+    }
     @Override
     public PersistenceAdapter setDefaultPersistenceAdapter(BrokerService broker) throws IOException {
         broker.setKeepDurableSubsActive(keepDurableSubsActive);
@@ -78,6 +109,8 @@ public class DurableSubscriptionOfflineBrowseRemoveTest extends DurableSubscript
     @Override
     protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://" + getName(true));
+        connectionFactory.setUserName(IDENTITY);
+        connectionFactory.setPassword(IDENTITY);
         connectionFactory.setWatchTopicAdvisories(false);
         return connectionFactory;
     }

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTestBase.java

@@ -130,10 +130,15 @@ public abstract class DurableSubscriptionOfflineTestBase {
             // have lots of journal files
             ((KahaDBPersistenceAdapter)broker.getPersistenceAdapter()).setJournalMaxFileLength(journalMaxFileLength);
         }
+
+        configurePlugins(broker);
         broker.start();
         broker.waitUntilStarted();
     }
 
+    public void configurePlugins(BrokerService broker) throws Exception {
+    }
+
     protected void destroyBroker() throws Exception {
         if (broker != null)
             broker.stop();