[AMQP 1.0 JMS] Thread deadlock when closing connection

Occasional thread deadlock observed.

In the scenario being tested, a bunch of session/producer pairs are being created over the same connection. Connection recovery logic is triggered off the connection’s exception listener. In the event of a fatal exception for the connection, the connection is teared down cleanly, iterating through all the associated sessions and producers and explicitly closing them before stopping and then closing the connection.

The pseudo-code his recovery logic goes like this:
. . .
Error occurs that signals connection must be closed and rebuilt
. . .
Wait for ReentrantReadWriteLock.write lock so no other threads are manipulating connections/sessions/producers, the publishers use read locks
Close all producer related to connection with producer.close()
Close all sessions related to connection with session.close()
Stop all connections with connection.stop()
Close all connections with connection.close()
DEADLOCKS OCCUR
Release write lock
. . .
Acquire ReentrantReadWriteLock.write lock
connection = factory.createConnection();
connection.setExceptionListener(this);
connection.start();
Release write lock
. . .

The following deadlock is observed(analyzed via JStack):

Found one Java-level deadlock:
=============================
"Thread-133":
waiting to lock monitor 0x00007f48d41ec840 (object 0x0000000789c3c3b0, a java.lang.Object),
which is held by "ConnectionManagementAndRecoveryWorker-id:0"
"ConnectionManagementAndRecoveryWorker-id:0":
waiting to lock monitor 0x00007f48440d9970 (object 0x0000000789c21de8, a org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint),
which is held by "Thread-133"

Java stack information for the threads listed above:
===================================================
"Thread-133":
at org.apache.qpid.amqp_1_0.jms.impl.ConnectionImpl.removeSession(ConnectionImpl.java:221)

• waiting to lock <0x0000000789c3c3b0> (a java.lang.Object)
  at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl.close(SessionImpl.java:274)
  at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl$1.remoteEnd(SessionImpl.java:106)
  at org.apache.qpid.amqp_1_0.transport.SessionEndpoint.end(SessionEndpoint.java:166)
• locked <0x0000000789c21de8> (a org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint)
  at org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint.receiveEnd(ConnectionEndpoint.java:544)
• locked <0x0000000789c21de8> (a org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint)
  at org.apache.qpid.amqp_1_0.type.transport.End.invoke(End.java:75)
  at org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint.receive(ConnectionEndpoint.java:684)
• locked <0x0000000789c21de8> (a org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint)
  at org.apache.qpid.amqp_1_0.framing.FrameHandler.parse(FrameHandler.java:242)
  at org.apache.qpid.amqp_1_0.framing.ConnectionHandler.parse(ConnectionHandler.java:70)
  at org.apache.qpid.amqp_1_0.client.Connection.doRead(Connection.java:396)
  at org.apache.qpid.amqp_1_0.client.Connection.access$000(Connection.java:47)
  at org.apache.qpid.amqp_1_0.client.Connection$2.run(Connection.java:259)
  at java.lang.Thread.run(Thread.java:724)
  "ConnectionManagementAndRecoveryWorker-id:0":
  at org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint.close(ConnectionEndpoint.java:721)
• waiting to lock <0x0000000789c21de8> (a org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint)
  at org.apache.qpid.amqp_1_0.client.Connection.close(Connection.java:419)
  at org.apache.qpid.amqp_1_0.jms.impl.ConnectionImpl.close(ConnectionImpl.java:359)
• locked <0x0000000789c3c3b0> (a java.lang.Object)
  at com.customer.stream.transport.amqp.qpid.QpidConnectionWrapper.close(QpidConnectionWrapper.java:296)
  at com.customer.stream.transport.amqp.qpid.QpidConnectionWrapper$ConnectionManagementAndRecoveryWorker.closeAndReopenConnectionWithExponentialBackoffError(QpidConnectionWrapper.java:1208)
  at com.customer.stream.transport.amqp.qpid.QpidConnectionWrapper$ConnectionManagementAndRecoveryWorker.manageConnection(QpidConnectionWrapper.java:1159)
  at com.customer.stream.transport.amqp.qpid.QpidConnectionWrapper$ConnectionManagementAndRecoveryWorker.run(QpidConnectionWrapper.java:1098)
  at java.lang.Thread.run(Thread.java:724)