Fix Impossible Cast issues in MemoryTopicSub:

- recoverSubscription()
-- map is defined as LinkedHashMap<MessageId, Message>
-- msg is defined as <map> entry.getValue() so must be a Message
-- condition if (msg.getClass() == MessageId.class) could never be true
-- no need to cast at all when using generics

- recoverNextMessages()
-- basically same code copy/pasted so same fix

Removed 2 conditions from ServerSessionPoolImpl that would result in impossible casts. Conditions removed were trying to cast ActiveMQQueueSession and ActiveMQTopicSession to ActiveMQSession which is illegal.

Since it isn't obvious what to do if you get an ActiveMQQueueSession or ActiveMQTopicSession from getServerSession() I make it fall back to the else condition which raises an async exception. This is better than getting a ClassCastException at runtime.

Remove impossible cast in MemoryMessageStore

activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java

@@ -108,12 +108,8 @@ public class MemoryMessageStore extends AbstractMessageStore {
         // here
         synchronized (messageTable) {
             for (Iterator<Message> iter = messageTable.values().iterator(); iter.hasNext();) {
-                Object msg = iter.next();
-                if (msg.getClass() == MessageId.class) {
-                    listener.recoverMessageReference((MessageId)msg);
-                } else {
-                    listener.recoverMessage((Message)msg);
-                }
+                Message msg = iter.next();
+                listener.recoverMessage(msg);
             }
         }
     }

activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryTopicSub.java

@@ -72,12 +72,8 @@ class MemoryTopicSub {
     synchronized void recoverSubscription(MessageRecoveryListener listener) throws Exception {
         for (Iterator<Entry<MessageId, Message>> iter = map.entrySet().iterator(); iter.hasNext();) {
             Entry<MessageId, Message> entry = iter.next();
-            Object msg = entry.getValue();
-            if (msg.getClass() == MessageId.class) {
-                listener.recoverMessageReference((MessageId)msg);
-            } else {
-                listener.recoverMessage((Message)msg);
-            }
+            Message msg = entry.getValue();
+            listener.recoverMessage(msg);
         }
     }
 
@@ -91,13 +87,9 @@ class MemoryTopicSub {
             Entry<MessageId, Message> entry = iter.next();
             if (pastLackBatch) {
                 count++;
-                Object msg = entry.getValue();
-                lastId = (MessageId)entry.getKey();
-                if (msg.getClass() == MessageId.class) {
-                    listener.recoverMessageReference((MessageId)msg);
-                } else {
-                    listener.recoverMessage((Message)msg);
-                }
+                Message msg = entry.getValue();
+                lastId = entry.getKey();
+                listener.recoverMessage(msg);
             } else {
                 pastLackBatch = entry.getKey().equals(lastBatch);
             }

activemq-ra/src/main/java/org/apache/activemq/ra/ServerSessionPoolImpl.java

@@ -265,10 +265,6 @@ public class ServerSessionPoolImpl implements ServerSessionPool {
         ActiveMQSession session = null;
         if (s instanceof ActiveMQSession) {
             session = (ActiveMQSession) s;
-        } else if (s instanceof ActiveMQQueueSession) {
-            session = (ActiveMQSession) s;
-        } else if (s instanceof ActiveMQTopicSession) {
-            session = (ActiveMQSession) s;
         } else {
             activeMQAsfEndpointWorker.getConnection()
                     .onAsyncException(new JMSException(