mirror of https://github.com/apache/activemq.git
KahaDB Recovery can experience a dangling transaction when prepare and commit occur on different pagefiles.
Signed-off-by: gtully <gary.tully@gmail.com>
This commit is contained in:
jgoodyear
gtully
parent
ac1e709dc4
commit
99db9ef301
|
|
@ -1408,6 +1408,9 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
|
|||
} finally {
|
||||
indexLock.writeLock().unlock();
|
||||
}
|
||||
for (Operation op: inflightTx) {
|
||||
recordAckMessageReferenceLocation(location, op.getLocation());
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
|
|
@ -1417,6 +1420,9 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
|
|||
List<Operation> tx = inflightTransactions.remove(key);
|
||||
if (tx != null) {
|
||||
preparedTransactions.put(key, tx);
|
||||
for (Operation op: tx) {
|
||||
recordAckMessageReferenceLocation(location, op.getLocation());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1431,6 +1437,11 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
|
|||
updates = preparedTransactions.remove(key);
|
||||
}
|
||||
}
|
||||
if (updates != null) {
|
||||
for(Operation op : updates) {
|
||||
recordAckMessageReferenceLocation(location, op.getLocation());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected void process(KahaRewrittenDataFileCommand command, Location location) throws IOException {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,360 @@
|
|||
package org.apache.activemq.bugs;
|
||||
|
||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
import org.apache.activemq.ActiveMQXAConnection;
|
||||
import org.apache.activemq.ActiveMQXAConnectionFactory;
|
||||
import org.apache.activemq.broker.BrokerFactory;
|
||||
import org.apache.activemq.broker.BrokerService;
|
||||
import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
|
||||
import org.apache.activemq.broker.jmx.BrokerViewMBean;
|
||||
import org.apache.activemq.broker.jmx.DestinationViewMBean;
|
||||
import org.apache.activemq.broker.jmx.PersistenceAdapterViewMBean;
|
||||
import org.apache.activemq.broker.jmx.QueueViewMBean;
|
||||
import org.apache.activemq.broker.jmx.RecoveredXATransactionViewMBean;
|
||||
import org.apache.activemq.command.TransactionId;
|
||||
import org.apache.activemq.command.XATransactionId;
|
||||
import org.apache.activemq.util.JMXSupport;
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import javax.jms.*;
|
||||
import javax.management.MalformedObjectNameException;
|
||||
import javax.management.ObjectName;
|
||||
import javax.transaction.xa.XAException;
|
||||
import javax.transaction.xa.XAResource;
|
||||
import javax.transaction.xa.Xid;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.File;
|
||||
import java.io.FileNotFoundException;
|
||||
import java.io.IOException;
|
||||
import java.io.PrintWriter;
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import java.net.URI;
|
||||
import java.util.Random;
|
||||
|
||||
import static javax.transaction.xa.XAResource.*;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
public class AMQ7067Test {
|
||||
|
||||
protected static Random r = new Random();
|
||||
final static String WIRE_LEVEL_ENDPOINT = "tcp://localhost:61616";
|
||||
protected BrokerService broker;
|
||||
protected ActiveMQXAConnection connection;
|
||||
protected XASession xaSession;
|
||||
protected XAResource xaRes;
|
||||
private final String xbean = "xbean:";
|
||||
private final String confBase = "src/test/resources/org/apache/activemq/bugs/amq7067";
|
||||
|
||||
|
||||
private static final ActiveMQXAConnectionFactory ACTIVE_MQ_CONNECTION_FACTORY;
|
||||
private static final ActiveMQConnectionFactory ACTIVE_MQ_NON_XA_CONNECTION_FACTORY;
|
||||
|
||||
static {
|
||||
ACTIVE_MQ_CONNECTION_FACTORY = new ActiveMQXAConnectionFactory(WIRE_LEVEL_ENDPOINT);
|
||||
ACTIVE_MQ_NON_XA_CONNECTION_FACTORY = new ActiveMQConnectionFactory(WIRE_LEVEL_ENDPOINT);
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setup() throws Exception {
|
||||
deleteData(new File("target/data"));
|
||||
createBroker();
|
||||
}
|
||||
|
||||
@After
|
||||
public void shutdown() throws Exception {
|
||||
broker.stop();
|
||||
}
|
||||
|
||||
public void setupXAConnection() throws Exception {
|
||||
connection = (ActiveMQXAConnection) ACTIVE_MQ_CONNECTION_FACTORY.createXAConnection();
|
||||
connection.start();
|
||||
xaSession = connection.createXASession();
|
||||
xaRes = xaSession.getXAResource();
|
||||
}
|
||||
|
||||
private void createBroker() throws Exception {
|
||||
broker = new BrokerService();
|
||||
broker = BrokerFactory.createBroker(xbean + confBase + "/activemq.xml");
|
||||
broker.start();
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testAMQ7067XAcommit() throws Exception {
|
||||
|
||||
PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
|
||||
setupXAConnection();
|
||||
|
||||
Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
|
||||
createDanglingTransaction(xaRes, xaSession, holdKahaDb);
|
||||
|
||||
MessageProducer holdKahaDbProducer = xaSession.createProducer(holdKahaDb);
|
||||
|
||||
XATransactionId txid = createXATransaction();
|
||||
System.out.println("****** create new txid = " + txid);
|
||||
xaRes.start(txid, TMNOFLAGS);
|
||||
|
||||
TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", 10));
|
||||
holdKahaDbProducer.send(helloMessage);
|
||||
xaRes.end(txid, TMSUCCESS);
|
||||
xaRes.prepare(txid);
|
||||
|
||||
Queue queue = xaSession.createQueue("test");
|
||||
|
||||
produce(xaRes, xaSession, queue, 100, 512 * 1024);
|
||||
xaRes.commit(txid, false);
|
||||
produce(xaRes, xaSession, queue, 100, 512 * 1024);
|
||||
|
||||
((org.apache.activemq.broker.region.Queue) broker.getRegionBroker().getDestinationMap().get(queue)).purge();
|
||||
|
||||
Xid[] xids = xaRes.recover(TMSTARTRSCAN);
|
||||
|
||||
//Should be 1 since we have only 1 prepared
|
||||
assertEquals(1, xids.length);
|
||||
connection.close();
|
||||
|
||||
broker.stop();
|
||||
broker.waitUntilStopped();
|
||||
createBroker();
|
||||
|
||||
setupXAConnection();
|
||||
xids = xaRes.recover(TMSTARTRSCAN);
|
||||
|
||||
// THIS SHOULD NOT FAIL AS THERE SHOUL DBE ONLY 1 TRANSACTION!
|
||||
assertEquals(1, xids.length);
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAMQ7067XArollback() throws Exception {
|
||||
|
||||
PersistenceAdapterViewMBean kahadbView = getProxyToPersistenceAdapter(broker.getPersistenceAdapter().toString());
|
||||
setupXAConnection();
|
||||
|
||||
Queue holdKahaDb = xaSession.createQueue("holdKahaDb");
|
||||
createDanglingTransaction(xaRes, xaSession, holdKahaDb);
|
||||
|
||||
MessageProducer holdKahaDbProducer = xaSession.createProducer(holdKahaDb);
|
||||
|
||||
XATransactionId txid = createXATransaction();
|
||||
System.out.println("****** create new txid = " + txid);
|
||||
xaRes.start(txid, TMNOFLAGS);
|
||||
|
||||
TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", 10));
|
||||
holdKahaDbProducer.send(helloMessage);
|
||||
xaRes.end(txid, TMSUCCESS);
|
||||
xaRes.prepare(txid);
|
||||
|
||||
Queue queue = xaSession.createQueue("test");
|
||||
|
||||
produce(xaRes, xaSession, queue, 100, 512 * 1024);
|
||||
xaRes.rollback(txid);
|
||||
produce(xaRes, xaSession, queue, 100, 512 * 1024);
|
||||
|
||||
((org.apache.activemq.broker.region.Queue) broker.getRegionBroker().getDestinationMap().get(queue)).purge();
|
||||
|
||||
Xid[] xids = xaRes.recover(TMSTARTRSCAN);
|
||||
|
||||
//Should be 1 since we have only 1 prepared
|
||||
assertEquals(1, xids.length);
|
||||
connection.close();
|
||||
|
||||
broker.stop();
|
||||
broker.waitUntilStopped();
|
||||
createBroker();
|
||||
|
||||
setupXAConnection();
|
||||
xids = xaRes.recover(TMSTARTRSCAN);
|
||||
|
||||
// THIS SHOULD NOT FAIL AS THERE SHOULD BE ONLY 1 TRANSACTION!
|
||||
assertEquals(1, xids.length);
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAMQ7067commit() throws Exception {
|
||||
final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
|
||||
connection.start();
|
||||
|
||||
Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
|
||||
Queue holdKahaDb = session.createQueue("holdKahaDb");
|
||||
MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
|
||||
TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
|
||||
holdKahaDbProducer.send(helloMessage);
|
||||
Queue queue = session.createQueue("test");
|
||||
produce(connection, queue, 100, 512*1024);
|
||||
session.commit();
|
||||
produce(connection, queue, 100, 512*1024);
|
||||
|
||||
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
|
||||
purgeQueue(queue.getQueueName());
|
||||
Thread.sleep(10000);
|
||||
|
||||
curruptIndexFile(getDataDirectory());
|
||||
|
||||
|
||||
while(true) {
|
||||
try {
|
||||
Thread.sleep(10000);
|
||||
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
|
||||
break;
|
||||
} catch (Exception ex) {
|
||||
System.out.println(ex.getMessage());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
connection.close();
|
||||
|
||||
// THIS SHOULD NOT FAIL AS THERE SHOULD BE ONLY 1 TRANSACTION!
|
||||
assertEquals(1, getQueueSize(holdKahaDb.getQueueName()));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAMQ7067rollback() throws Exception {
|
||||
final Connection connection = ACTIVE_MQ_NON_XA_CONNECTION_FACTORY.createConnection();
|
||||
connection.start();
|
||||
|
||||
Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
|
||||
Queue holdKahaDb = session.createQueue("holdKahaDb");
|
||||
MessageProducer holdKahaDbProducer = session.createProducer(holdKahaDb);
|
||||
TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", 10));
|
||||
holdKahaDbProducer.send(helloMessage);
|
||||
Queue queue = session.createQueue("test");
|
||||
produce(connection, queue, 100, 512*1024);
|
||||
session.rollback();
|
||||
produce(connection, queue, 100, 512*1024);
|
||||
|
||||
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
|
||||
purgeQueue(queue.getQueueName());
|
||||
Thread.sleep(10000);
|
||||
|
||||
curruptIndexFile(getDataDirectory());
|
||||
|
||||
|
||||
while(true) {
|
||||
try {
|
||||
Thread.sleep(10000);
|
||||
System.out.println(String.format("QueueSize %s: %d", holdKahaDb.getQueueName(), getQueueSize(holdKahaDb.getQueueName())));
|
||||
break;
|
||||
} catch (Exception ex) {
|
||||
System.out.println(ex.getMessage());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
connection.close();
|
||||
|
||||
// THIS SHOULD NOT FAIL AS THERE SHOULD ZERO TRANSACTION!
|
||||
assertEquals(0, getQueueSize(holdKahaDb.getQueueName()));
|
||||
}
|
||||
|
||||
protected static void createDanglingTransaction(XAResource xaRes, XASession xaSession, Queue queue) throws JMSException, IOException, XAException {
|
||||
MessageProducer producer = xaSession.createProducer(queue);
|
||||
XATransactionId txId = createXATransaction();
|
||||
xaRes.start(txId, TMNOFLAGS);
|
||||
|
||||
TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("dangler", 10));
|
||||
producer.send(helloMessage);
|
||||
xaRes.end(txId, TMSUCCESS);
|
||||
xaRes.prepare(txId);
|
||||
System.out.println("****** createDanglingTransaction txId = " + txId);
|
||||
}
|
||||
|
||||
protected static void produce(XAResource xaRes, XASession xaSession, Queue queue, int messageCount, int messageSize) throws JMSException, IOException, XAException {
|
||||
MessageProducer producer = xaSession.createProducer(queue);
|
||||
|
||||
for (int i = 0; i < messageCount; i++) {
|
||||
XATransactionId txid = createXATransaction();
|
||||
xaRes.start(txid, TMNOFLAGS);
|
||||
|
||||
TextMessage helloMessage = xaSession.createTextMessage(StringUtils.repeat("a", messageSize));
|
||||
producer.send(helloMessage);
|
||||
xaRes.end(txid, TMSUCCESS);
|
||||
xaRes.prepare(txid);
|
||||
xaRes.commit(txid, true);
|
||||
}
|
||||
}
|
||||
|
||||
protected static void produce(Connection connection, Queue queue, int messageCount, int messageSize) throws JMSException, IOException, XAException {
|
||||
Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
|
||||
MessageProducer producer = session.createProducer(queue);
|
||||
|
||||
for (int i = 0; i < messageCount; i++) {
|
||||
TextMessage helloMessage = session.createTextMessage(StringUtils.repeat("a", messageSize));
|
||||
producer.send(helloMessage);
|
||||
session.commit();
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
protected static XATransactionId createXATransaction() throws IOException {
|
||||
ByteArrayOutputStream baos = new ByteArrayOutputStream();
|
||||
DataOutputStream os = new DataOutputStream(baos);
|
||||
os.writeLong(r.nextInt());
|
||||
os.close();
|
||||
byte[] bs = baos.toByteArray();
|
||||
|
||||
XATransactionId xid = new XATransactionId();
|
||||
xid.setBranchQualifier(bs);
|
||||
xid.setGlobalTransactionId(bs);
|
||||
xid.setFormatId(55);
|
||||
return xid;
|
||||
}
|
||||
|
||||
private RecoveredXATransactionViewMBean getProxyToPreparedTransactionViewMBean(TransactionId xid) throws MalformedObjectNameException, JMSException {
|
||||
|
||||
ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,transactionType=RecoveredXaTransaction,xid=" +
|
||||
JMXSupport.encodeObjectNamePart(xid.toString()));
|
||||
RecoveredXATransactionViewMBean proxy = (RecoveredXATransactionViewMBean) broker.getManagementContext().newProxyInstance(objectName,
|
||||
RecoveredXATransactionViewMBean.class, true);
|
||||
return proxy;
|
||||
}
|
||||
|
||||
private PersistenceAdapterViewMBean getProxyToPersistenceAdapter(String name) throws MalformedObjectNameException, JMSException {
|
||||
return (PersistenceAdapterViewMBean) broker.getManagementContext().newProxyInstance(
|
||||
BrokerMBeanSupport.createPersistenceAdapterName(broker.getBrokerObjectName().toString(), name),
|
||||
PersistenceAdapterViewMBean.class, true);
|
||||
}
|
||||
|
||||
private void deleteData(File file) throws Exception {
|
||||
String[] entries = file.list();
|
||||
if (entries == null) return;
|
||||
for (String s : entries) {
|
||||
File currentFile = new File(file.getPath(), s);
|
||||
if (currentFile.isDirectory()) {
|
||||
deleteData(currentFile);
|
||||
}
|
||||
currentFile.delete();
|
||||
}
|
||||
file.delete();
|
||||
}
|
||||
|
||||
private long getQueueSize(final String queueName) throws MalformedObjectNameException {
|
||||
ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=" + JMXSupport.encodeObjectNamePart(queueName));
|
||||
DestinationViewMBean proxy = (DestinationViewMBean) broker.getManagementContext().newProxyInstance(objectName, DestinationViewMBean.class, true);
|
||||
return proxy.getQueueSize();
|
||||
}
|
||||
|
||||
private void purgeQueue(final String queueName) throws MalformedObjectNameException, Exception {
|
||||
ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=" + JMXSupport.encodeObjectNamePart(queueName));
|
||||
QueueViewMBean proxy = (QueueViewMBean) broker.getManagementContext().newProxyInstance(objectName, QueueViewMBean.class, true);
|
||||
proxy.purge();
|
||||
}
|
||||
|
||||
private String getDataDirectory() throws MalformedObjectNameException {
|
||||
ObjectName objectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost");
|
||||
BrokerViewMBean proxy = (BrokerViewMBean) broker.getManagementContext().newProxyInstance(objectName, BrokerViewMBean.class, true);
|
||||
return proxy.getDataDirectory();
|
||||
}
|
||||
|
||||
protected static void curruptIndexFile(final String dataPath) throws FileNotFoundException, UnsupportedEncodingException {
|
||||
PrintWriter writer = new PrintWriter(String.format("%s/kahadb/db.data", dataPath), "UTF-8");
|
||||
writer.println("asdasdasd");
|
||||
writer.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,111 @@
|
|||
<!--
|
||||
Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
contributor license agreements. See the NOTICE file distributed with
|
||||
this work for additional information regarding copyright ownership.
|
||||
The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
(the "License"); you may not use this file except in compliance with
|
||||
the License. You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
-->
|
||||
<!-- START SNIPPET: example -->
|
||||
<beans
|
||||
xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
|
||||
|
||||
<!--
|
||||
The <broker> element is used to configure the ActiveMQ broker.
|
||||
-->
|
||||
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="target/data">
|
||||
|
||||
<destinationPolicy>
|
||||
<policyMap>
|
||||
<policyEntries>
|
||||
<policyEntry topic=">" >
|
||||
<!-- The constantPendingMessageLimitStrategy is used to prevent
|
||||
slow topic consumers to block producers and affect other consumers
|
||||
by limiting the number of messages that are retained
|
||||
For more information, see:
|
||||
|
||||
http://activemq.apache.org/slow-consumer-handling.html
|
||||
|
||||
-->
|
||||
<pendingMessageLimitStrategy>
|
||||
<constantPendingMessageLimitStrategy limit="1000"/>
|
||||
</pendingMessageLimitStrategy>
|
||||
</policyEntry>
|
||||
</policyEntries>
|
||||
</policyMap>
|
||||
</destinationPolicy>
|
||||
|
||||
|
||||
<!--
|
||||
The managementContext is used to configure how ActiveMQ is exposed in
|
||||
JMX. By default, ActiveMQ uses the MBean server that is started by
|
||||
the JVM. For more information, see:
|
||||
|
||||
http://activemq.apache.org/jmx.html
|
||||
-->
|
||||
<managementContext>
|
||||
<managementContext createConnector="false"/>
|
||||
</managementContext>
|
||||
|
||||
<!--
|
||||
Configure message persistence for the broker. The default persistence
|
||||
mechanism is the KahaDB store (identified by the kahaDB tag).
|
||||
For more information, see:
|
||||
|
||||
http://activemq.apache.org/persistence.html
|
||||
-->
|
||||
<persistenceAdapter>
|
||||
<kahaDB directory="target/data/kahadb" />
|
||||
</persistenceAdapter>
|
||||
|
||||
|
||||
<!--
|
||||
The systemUsage controls the maximum amount of space the broker will
|
||||
use before disabling caching and/or slowing down producers. For more information, see:
|
||||
http://activemq.apache.org/producer-flow-control.html
|
||||
-->
|
||||
<systemUsage>
|
||||
<systemUsage>
|
||||
<memoryUsage>
|
||||
<memoryUsage percentOfJvmHeap="70" />
|
||||
</memoryUsage>
|
||||
<storeUsage>
|
||||
<storeUsage limit="100 gb"/>
|
||||
</storeUsage>
|
||||
<tempUsage>
|
||||
<tempUsage limit="50 gb"/>
|
||||
</tempUsage>
|
||||
</systemUsage>
|
||||
</systemUsage>
|
||||
|
||||
<!--
|
||||
The transport connectors expose ActiveMQ over a given protocol to
|
||||
clients and other brokers. For more information, see:
|
||||
|
||||
http://activemq.apache.org/configuring-transports.html
|
||||
-->
|
||||
<transportConnectors>
|
||||
<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
|
||||
<transportConnector name="openwire" uri="tcp://localhost:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
|
||||
</transportConnectors>
|
||||
|
||||
<!-- destroy the spring context on shutdown to stop jetty -->
|
||||
<shutdownHooks>
|
||||
<bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
|
||||
</shutdownHooks>
|
||||
|
||||
</broker>
|
||||
|
||||
</beans>
|
||||
<!-- END SNIPPET: example -->
|
||||
Loading…
Reference in New Issue