@Test
public void testConnectionRestoredExpectRecreateSession() throws JMSException {

    final Target target = TestConstants.Targets.TWIN_TARGET;
    final JmsMessageConsumer recoveredConsumer = Mockito.mock(JmsMessageConsumer.class);
    final JmsSession newSession = Mockito.mock(JmsSession.class, withSettings().name("recoveredSession"));
    when(mockSession.isClosed()).thenReturn(true); // existing session was closed
    doReturn(mockSession) // initial session
            .doReturn(newSession) // recovered session
            .when(mockConnection)
            .createSession(anyInt());
    prepareSession(newSession, recoveredConsumer);
    new TestKit(actorSystem) {{
        final Props props =
                AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection);
        final ActorRef amqpClientActor = actorSystem.actorOf(props);
        // connect
        amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef());
        expectMsg(CONNECTED_SUCCESS);
        // capture connection listener to close session later
        final ArgumentCaptor<JmsConnectionListener> captor = ArgumentCaptor.forClass(JmsConnectionListener.class);
        verify(mockConnection).addConnectionListener(captor.capture());
        final JmsConnectionListener jmsConnectionListener = captor.getValue();
        // verify everything is setup correctly by publishing an event
        sendThingEventAndExpectPublish(amqpClientActor, target, () -> getProducerForAddress(target.getAddress()));
        // verify message is consumed and forwarded to concierge
        consumeMockMessage(mockConsumer);
        expectMsgClass(Command.class);
        // now close session
        jmsConnectionListener.onConnectionRestored(URI.create("amqp://broker:5671"));
        verify(mockConnection, timeout(2000).times(2)).createSession(anyInt());
        // close is called on old session
        verify(mockSession, timeout(2000).times(2)).close();
        // verify publishing an event works with new session/producer
        sendThingEventAndExpectPublish(amqpClientActor, target, () -> getProducerForAddress(target.getAddress()));
        // verify message is consumed with newly created consumer
        consumeMockMessage(recoveredConsumer);
        expectMsgClass(Command.class);
    }};
}