[
https://issues.apache.org/jira/browse/AMQ-3213?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16943339#comment-16943339
]
Drobyshev Maksim commented on AMQ-3213:
---------------------------------------
[~tabish] hello, we got the same issue, but we use 5.15.10 version.
> failover doesn't reconnect after broker restart
> -----------------------------------------------
>
> Key: AMQ-3213
> URL: https://issues.apache.org/jira/browse/AMQ-3213
> Project: ActiveMQ
> Issue Type: Bug
> Components: Broker
> Affects Versions: 5.4.2
> Environment: Windows 7 x64, jdk 1.6.0_21
> Reporter: Ivan Shcheklein
> Priority: Major
> Fix For: 5.6.0
>
>
> I have a network of three brokers: two embedded (publisher and subscriber)
> and one stand alone remote. Embedded brokers connect to the remote with
> duplex static failover connections. Everything works fine if remote broker is
> not restarted. After the remote broker is restarted embedded brokers fail to
> reestablish duplex bridges. Subscriber doesn't receive any messages and in
> logs I have:
> *Subscriber worker:*
> {code}
> Received 14-th message.
> Received 15-th message.
> Received 16-th message.
> 2011:03:09 18:19:08,276 [WARN ]
> org.apache.activemq.transport.failover.FailoverTransport - Transport
> (localhost/127.0.0.1:61616) failed to tcp://localhost:61616 , attempting to
> automatically reconnect due to: java.net.SocketException: Connection reset
> 2011:03:09 18:19:08,276 [INFO ]
> org.apache.activemq.network.DemandForwardingBridgeSupport - Outbound
> transport to remote interrupted.
> 2011:03:09 18:19:22,426 [INFO ]
> org.apache.activemq.network.DemandForwardingBridgeSupport - Network
> connection between vm://local1#0 and tcp://localhost:61616(remote) has been
> established.
> 2011:03:09 18:19:22,429 [INFO ]
> org.apache.activemq.network.DemandForwardingBridgeSupport - Outbound
> transport to remote resumed
> 2011:03:09 18:19:22,429 [INFO ]
> org.apache.activemq.transport.failover.FailoverTransport - Successfully
> reconnected to tcp://localhost:61616
> 2011:03:09 18:19:22,455 [WARN ]
> org.apache.activemq.broker.TransportConnection - Unexpected extra broker info
> command received: BrokerInfo {commandId = 1929, responseRequired = false,
> brokerId = ID:Air-17262-1299683959982-0:1, brokerURL =
> tcp://validation.sls.microsoft.com:61616, slaveBroker = false, masterBroker =
> false, faultTolerantConfiguration = false, networkConnection = false,
> duplexConnection = false, peerBrokerInfos = [], brokerName = remote,
> connectionId = 0, brokerUploadUrl = null, networkProperties = null}
> {code}
> *Publisher worker:*
> {code}
> Sending 28-th message
> Sending 29-th message
> Sending 30-th message
> 2011:03:09 18:19:22,430 [INFO ]
> org.apache.activemq.network.DemandForwardingBridgeSupport - Network
> connection between vm://local#0 and tcp://localhost:61616(remote) has been
> established.
> 2011:03:09 18:19:22,435 [INFO ]
> org.apache.activemq.network.DemandForwardingBridgeSupport - Outbound
> transport to remote resumed
> 2011:03:09 18:19:22,435 [INFO ]
> org.apache.activemq.transport.failover.FailoverTransport - Successfully
> reconnected to tcp://localhost:61616
> 2011:03:09 18:19:22,469 [WARN ]
> org.apache.activemq.broker.TransportConnection - Unexpected extra broker info
> command received: BrokerInfo {commandId = 1911, responseRequired = false,
> brokerId = ID:Air-17262-1299683959982-0:1, brokerURL =
> tcp://validation.sls.microsoft.com:61616, slaveBroker = false, masterBroker =
> false, faultTolerantConfiguration = false, networkConnection = false,
> duplexConnection = false, peerBrokerInfos = [], brokerName = remote,
> connectionId = 0, brokerUploadUrl = null, networkProperties = null}
> Sending 31-th message
> Sending 32-th message
> Sending 33-th message
> {code}
> *Sample code to reproduce this issue:*
> {code}
> import org.apache.activemq.ActiveMQConnectionFactory;
> import org.apache.activemq.broker.BrokerService;
> import org.apache.activemq.broker.TransportConnector;
> import org.apache.activemq.network.NetworkConnector;
> import javax.jms.*;
> import java.net.URI;
> public class JmsTester implements MessageListener {
> private static final int size = 256;
> private static byte[] payload;
> static {
> char[] DATA = "abcdefghijklmnopqrstuvwxyz".toCharArray();
> payload = new byte[size];
> for (int i = 0; i < size; i++) {
> payload[i] = (byte)DATA[i % DATA.length];
> }
> }
> public static void main(String[] args) throws Exception {
> if(System.getProperty("jms.mode").equals("publisher"))
> publisher();
> else if(System.getProperty("jms.mode").equals("broker"))
> broker();
> else
> subscriber();
> }
> private static void publisher() throws Exception {
> System.out.println("Starting publisher ...");
> BrokerService broker = new BrokerService();
> broker.setBrokerName("local");
> broker.setUseJmx(true);
> broker.setPersistent(true);
> NetworkConnector nc =
> broker.addNetworkConnector("static:(failover:(tcp://localhost:61616))");
> nc.setDuplex(true);
> nc.setNetworkTTL(4);
> broker.start();
> ActiveMQConnectionFactory factory = new
> ActiveMQConnectionFactory("vm://local");
> Connection connection = factory.createConnection();
> Session session = connection.createSession(false,
> Session.DUPS_OK_ACKNOWLEDGE);
> Topic topic = session.createTopic("topictest.messages");
> MessageProducer publisher = session.createProducer(topic);
> publisher.setDeliveryMode(DeliveryMode.PERSISTENT);
> for (int i=0; i<1000; i++) {
> Thread.sleep(1000);
> System.out.println("Sending " + i + "-th message");
> BytesMessage msg = session.createBytesMessage();
> msg.setIntProperty("count", i);
> msg.writeBytes(payload);
> publisher.send(msg);
> }
> broker.stop();
> }
> private static void broker() throws Exception {
> System.out.println("Starting broker ...");
> BrokerService broker = new BrokerService();
> broker.setPersistent(true);
> broker.setBrokerName("remote");
> broker.setUseJmx(true);
> TransportConnector connector = new TransportConnector();
> connector.setUri(new URI("tcp://localhost:61616"));
>
> broker.addConnector(connector);
> broker.start();
> while(true) { }
> }
> private static void subscriber() throws Exception {
> System.out.println("Starting subscriber ...");
> BrokerService broker = new BrokerService();
> broker.setBrokerName("local1");
> broker.setUseJmx(true);
> broker.setPersistent(true);
> NetworkConnector nc =
> broker.addNetworkConnector("static:(failover:(tcp://localhost:61616))");
> nc.setDuplex(true);
> nc.setNetworkTTL(4);
> broker.start();
> ActiveMQConnectionFactory factory = new
> ActiveMQConnectionFactory("vm://local1");
> Connection connection = factory.createConnection();
> connection.setClientID("subscriber2");
> Session session = connection.createSession(false,
> Session.DUPS_OK_ACKNOWLEDGE);
> Topic topic = session.createTopic("topictest.messages");
> MessageConsumer consumer = session.createDurableSubscriber(topic,
> "subscriber2");
> JmsTester tester = new JmsTester();
> consumer.setMessageListener(tester);
> connection.start();
> while(true) {
> Thread.sleep(1000);
> }
> }
> @Override
> public void onMessage(Message message) {
> try {
> int count = message.getIntProperty("count");
> System.out.println("Received " + count + "-th message.");
> } catch (JMSException e) {
> e.printStackTrace();
> }
> }
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
