ARTEMIS-4575 Only start the consumers that were added

Change from forcing a session start cycle on each consumer add
event and start only those consumers that were added which will
trigger a prompt delivery action on each. The session should be
marked started on create to account for the remove of the start
on each consumer add event.

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java

@@ -982,7 +982,7 @@ public void addConsumer(ConsumerInfo info) throws Exception {
             return;
          }
 
-         amqSession.start();
+         consumersList.forEach((c) -> c.start());
 
          if (AdvisorySupport.isAdvisoryTopic(info.getDestination())) {
             //advisory for temp destinations
@@ -1122,6 +1122,7 @@ public void addSessions(Set<SessionId> sessionSet) {
    public AMQSession addSession(SessionInfo ss) {
       AMQSession amqSession = new AMQSession(getState().getInfo(), ss, server, this, protocolManager, coreMessageObjectPools);
       amqSession.initialize();
+      amqSession.start();
 
       sessions.put(ss.getSessionId(), amqSession);
       sessionIdMap.put(amqSession.getCoreSession().getName(), ss.getSessionId());

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java

@@ -137,6 +137,14 @@ private static String convertOpenWireToActiveMQFilterString(final String selecto
       return filterString;
    }
 
+   public void start() {
+      if (serverConsumer == null) {
+         throw new IllegalStateException("Cannot start the AMQConsumer until it has been initialized");
+      }
+
+      serverConsumer.setStarted(true);
+   }
+
    public void init(SlowConsumerDetectionListener slowConsumerDetectionListener, long nativeId) throws Exception {
 
       SimpleString selector = info.getSelector() == null ? null : new SimpleString(convertOpenWireToActiveMQFilterString(info.getSelector()));