[jms] connection:forced leads to JMSException even though reconnect is enabled

diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
index 81c28855ef..888171227b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
@@ -86,6 +86,32 @@ public class AmqpFailoverEndpointDiscoveryTest extends FailoverTestBase {
       }
    }
 
+   @Test(timeout = 120000)
+   public void testReconnectWhileSendingWithAMQP() throws Exception {
+      JmsConnectionFactory factory = getJmsConnectionFactory();
+      try (Connection connection = factory.createConnection()) {
+         connection.start();
+         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         javax.jms.Queue queue = session.createQueue(ADDRESS.toString());
+         MessageProducer producer = session.createProducer(queue);
+         Thread t = new Thread(() -> {
+            try {
+               while(true) {
+                  System.out.println("sending message");
+                  producer.send(session.createTextMessage("hello before failover"));
+//                  Thread.sleep(1000);  // comment out to send messages quickly
+               }
+            } catch (Exception e ) {
+               e.printStackTrace();
+            }
+         });
+         t.start();
+         Thread.sleep(2000);  // simpler to read than actual synchronization
+         liveServer.crash(true, true);
+         Thread.sleep(2000);
+      }
+   }
+
    private JmsConnectionFactory getJmsConnectionFactory() {
       if (protocol == 0) {
          return new JmsConnectionFactory("failover:(amqp://localhost:61616)");

javax.jms.JMSException: Received error from remote peer without description [condition = amqp:connection:forced]
	at org.apache.qpid.jms.provider.amqp.AmqpSupport.convertToException(AmqpSupport.java:164)
	at org.apache.qpid.jms.provider.amqp.AmqpSupport.convertToException(AmqpSupport.java:117)
	at org.apache.qpid.jms.provider.amqp.AmqpAbstractResource.processRemoteClose(AmqpAbstractResource.java:262)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider.processUpdates(AmqpProvider.java:971)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider.access$1900(AmqpProvider.java:105)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider$17.run(AmqpProvider.java:854)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[...]
[1476077082:1] -> Transfer{handle=0, deliveryId=221, deliveryTag=\x00, messageFormat=0, settled=null, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=false} (179) "\x00Sp\xc0\x02\x01A\x00Sr\xc1)\x04\xa3\x0ex-opt-jms-destQ\x00\xa3\x12x-opt-jms-msg-typeQ\x05\x00Ss\xd0\x00\x00\x00\x5c\x00\x00\x00\x0a\xa11ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1-222@\xa1\x13FailoverTestAddress@@@@@@\x83\x00\x00\x01d\x12f\x06\xac\x00Sw\xa1\x15hello before failover"
[913923932:1] <- Transfer{handle=0, deliveryId=221, deliveryTag=\x00, messageFormat=0, settled=null, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=false} (179) "\x00Sp\xc0\x02\x01A\x00Sr\xc1)\x04\xa3\x0ex-opt-jms-destQ\x00\xa3\x12x-opt-jms-msg-typeQ\x05\x00Ss\xd0\x00\x00\x00\x5c\x00\x00\x00\x0a\xa11ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1-222@\xa1\x13FailoverTestAddress@@@@@@\x83\x00\x00\x01d\x12f\x06\xac\x00Sw\xa1\x15hello before failover"
[913923932:1] -> Disposition{role=RECEIVER, first=221, last=221, settled=true, state=Accepted{}, batchable=false}
[1476077082:1] <- Disposition{role=RECEIVER, first=221, last=221, settled=true, state=Accepted{}, batchable=false}
sending message
[1476077082:1] -> Transfer{handle=0, deliveryId=222, deliveryTag=\x00, messageFormat=0, settled=null, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=false} (179) "\x00Sp\xc0\x02\x01A\x00Sr\xc1)\x04\xa3\x0ex-opt-jms-destQ\x00\xa3\x12x-opt-jms-msg-typeQ\x05\x00Ss\xd0\x00\x00\x00\x5c\x00\x00\x00\x0a\xa11ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1-223@\xa1\x13FailoverTestAddress@@@@@@\x83\x00\x00\x01d\x12f\x06\xb0\x00Sw\xa1\x15hello before failover"
[913923932:0] -> Close{error=Error{condition=amqp:connection:forced, description='null', info=null}}
[913923932:1] <- Transfer{handle=0, deliveryId=222, deliveryTag=\x00, messageFormat=0, settled=null, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=false} (179) "\x00Sp\xc0\x02\x01A\x00Sr\xc1)\x04\xa3\x0ex-opt-jms-destQ\x00\xa3\x12x-opt-jms-msg-typeQ\x05\x00Ss\xd0\x00\x00\x00\x5c\x00\x00\x00\x0a\xa11ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1-223@\xa1\x13FailoverTestAddress@@@@@@\x83\x00\x00\x01d\x12f\x06\xb0\x00Sw\xa1\x15hello before failover"
[1476077082:0] <- Close{error=Error{condition=amqp:connection:forced, description='null', info=null}}
[1476077082:0] -> Close{error=null}
[913923932:0] <- Close{error=null}
javax.jms.JMSException: Received error from remote peer without description [condition = amqp:connection:forced]
	at org.apache.qpid.jms.provider.amqp.AmqpSupport.convertToException(AmqpSupport.java:164)
	at org.apache.qpid.jms.provider.amqp.AmqpSupport.convertToException(AmqpSupport.java:117)
	at org.apache.qpid.jms.provider.amqp.AmqpAbstractResource.processRemoteClose(AmqpAbstractResource.java:262)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider.processUpdates(AmqpProvider.java:971)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider.access$1900(AmqpProvider.java:105)
	at org.apache.qpid.jms.provider.amqp.AmqpProvider$17.run(AmqpProvider.java:854)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
[Thread-8 (ActiveMQ-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$5@343570b7)] 12:17:13,704 WARN  [org.apache.activemq.artemis.core.client] AMQ212037: Connection failure has been detected: AMQ119015: The connection was disconnected because of server shutdown [code=DISCONNECTED]
[Thread-6 (ActiveMQ-client-global-threads)] 12:17:13,704 WARN  [org.apache.activemq.artemis.core.client] AMQ212037: Connection failure has been detected: AMQ119015: The connection was disconnected because of server shutdown [code=DISCONNECTED]
[Thread-2] 12:17:13,731 INFO  [org.apache.activemq.artemis.core.server] AMQ221002: Apache ActiveMQ Artemis Message Broker version identity=AmqpFailoverEndpointDiscoveryTest/liveServer,version=2.7.0-SNAPSHOT [c1f0e25e-72e0-11e8-acbd-b6cd2f5fd8dd] stopped, uptime 4.323 seconds
[Thread-6 (ActiveMQ-client-global-threads)] 12:17:13,737 WARN  [org.apache.activemq.artemis.core.client] AMQ212004: Failed to connect to server.
[AMQ119000: Activation for server ActiveMQServerImpl::AmqpFailoverEndpointDiscoveryTest/backupServers] 12:17:13,831 INFO  [org.apache.activemq.artemis.core.server] AMQ221020: Started EPOLL Acceptor at localhost:61617 for protocols [CORE,MQTT,AMQP,STOMP,HORNETQ,OPENWIRE]
[AMQ119000: Activation for server ActiveMQServerImpl::AmqpFailoverEndpointDiscoveryTest/backupServers] 12:17:13,831 INFO  [org.apache.activemq.artemis.core.server] AMQ221010: Backup Server is now live
[737743497:0] -> Open{ containerId='ID:b8e3f0c3-0810-41d0-a087-c69de89f017f:1', hostname='localhost', maxFrameSize=1048576, channelMax=32767, idleTimeOut=30000, outgoingLocales=null, incomingLocales=null, offeredCapabilities=null, desiredCapabilities=[sole-connection-for-container, DELAYED_DELIVERY, ANONYMOUS-RELAY, SHARED-SUBS], properties={product=QpidJMS, version=0.32.0, platform=JVM: 1.8.0_172, 25.172-b11, Oracle Corporation, OS: Linux, 4.14.43, amd64}}
[1583552999:0] <- Open{ containerId='ID:b8e3f0c3-0810-41d0-a087-c69de89f017f:1', hostname='localhost', maxFrameSize=1048576, channelMax=32767, idleTimeOut=30000, outgoingLocales=null, incomingLocales=null, offeredCapabilities=null, desiredCapabilities=[sole-connection-for-container, DELAYED_DELIVERY, ANONYMOUS-RELAY, SHARED-SUBS], properties={product=QpidJMS, version=0.32.0, platform=JVM: 1.8.0_172, 25.172-b11, Oracle Corporation, OS: Linux, 4.14.43, amd64}}
[1583552999:0] -> Open{ containerId='localhost', hostname='null', maxFrameSize=131072, channelMax=65535, idleTimeOut=30000, outgoingLocales=null, incomingLocales=null, offeredCapabilities=[sole-connection-for-container, DELAYED_DELIVERY, SHARED-SUBS, ANONYMOUS-RELAY], desiredCapabilities=null, properties={product=apache-activemq-artemis, version=2.7.0-SNAPSHOT}}
[737743497:0] <- Open{ containerId='localhost', hostname='null', maxFrameSize=131072, channelMax=65535, idleTimeOut=30000, outgoingLocales=null, incomingLocales=null, offeredCapabilities=[sole-connection-for-container, DELAYED_DELIVERY, SHARED-SUBS, ANONYMOUS-RELAY], desiredCapabilities=null, properties={product=apache-activemq-artemis, version=2.7.0-SNAPSHOT}}
[737743497:0] -> Begin{remoteChannel=null, nextOutgoingId=1, incomingWindow=2047, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1583552999:0] <- Begin{remoteChannel=null, nextOutgoingId=1, incomingWindow=2047, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1583552999:0] -> Begin{remoteChannel=0, nextOutgoingId=1, incomingWindow=2147483647, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[737743497:0] <- Begin{remoteChannel=0, nextOutgoingId=1, incomingWindow=2147483647, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[737743497:1] -> Begin{remoteChannel=null, nextOutgoingId=1, incomingWindow=2047, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1583552999:1] <- Begin{remoteChannel=null, nextOutgoingId=1, incomingWindow=2047, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1583552999:1] -> Begin{remoteChannel=1, nextOutgoingId=1, incomingWindow=2147483647, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[737743497:1] <- Begin{remoteChannel=1, nextOutgoingId=1, incomingWindow=2147483647, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[737743497:1] -> Attach{name='qpid-jms:sender:ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1:FailoverTestAddress', handle=0, role=SENDER, sndSettleMode=UNSETTLED, rcvSettleMode=FIRST, source=Source{address='ID:bc37e33f-5706-4fe3-b953-c5be80e38b52:1:1:1', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=[amqp:accepted:list, amqp:rejected:list, amqp:released:list, amqp:modified:list], capabilities=null}, target=Target{address='FailoverTestAddress', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=[queue]}, unsettled=null, incompleteUnsettled=false, initialDeliveryCount=0, maxMessageSize=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[...]