Graceful shutdown for core bridge

Scenario:
1) Start 2 servers with configured core bridge from queue sourceQueue (server1) to queue targetQueue (server2)
2) Send 10 large messages to sourceQueue (to server1)
3) Receive 10 messages from targetQueue (from server2) a commit them.
4) Immediately cleanly shutdown server1

Result:
If step 4. (shutdown of server1) was executed fast enough then there are still messages in sourceQueue.

Investigation showed that messages were not acknowledged by bridge on server1 in time. From trace logs of server1, there is call BridgeImpl#sendAcknowledged(message) which should ack the message in the journal. This means that server2 send acknowledge that message was successfully received. But because server1 is shutting down, bridge is marked as not "active" and code which acknowledges the message is skipped. At the same time server2 thinks that messages were acknowledge and allows consumption of those messages from server2. Result of this is that after restart of server1, those message can be consumed from server1 by any receiver or resend to by core bridge server2.

The variable "active" was switched to false by:

15:36:40,062 main INFO  [org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl:374] mnovak - stop on bridge was called from: java.lang.Thread.getStackTrace(Thread.java:1552)
org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl.stop(BridgeImpl.java:369)
org.apache.activemq.artemis.core.server.cluster.ClusterManager.stop(ClusterManager.java:323)
org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.stopComponent(ActiveMQServerImpl.java:1115)
org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.stop(ActiveMQServerImpl.java:863)
org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.stop(ActiveMQServerImpl.java:813)
org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.internalStop(ActiveMQServerImpl.java:680)
org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.stop(ActiveMQServerImpl.java:673)

The idea is to allow to finish all deliveries in progress and stop all new delivery calls by bridge when BridgeImpl.stop() is called.