How to handle Activemq's max frame size exception with failover transport

Question

I am developing an application that uses activemq to exchange messages, some so big that I want to cancel then.

We work with the activemq failover transport with two ActiveMQ instances (master/slave). The broker itself has the 100mb framesize limit for messages.

The problem is: if I try to send a message bigger than 100mb the ActiveMQ server will shutdown the connection. At this point, the failover transport will try to reconnect and send the message again, creating an infinite loop.

The client logs the following:

2017-01-05 09:19:11.910  WARN 14680 --- [0.1:61616@57025] o.a.a.t.failover.FailoverTransport       : Transport (tcp://localhost:61616) failed , attempting to automatically reconnect: {}

java.io.EOFException: null
    at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.8.0_91]
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:267) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215) ~[activemq-client-5.13.4.jar:5.13.4]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]

2017-01-05 09:19:11.921  INFO 14680 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully reconnected to tcp://localhost:61616
2017-01-05 09:19:11.923  WARN 14680 --- [0.1:61616@57026] o.a.a.t.failover.FailoverTransport       : Transport (tcp://localhost:61616) failed , attempting to automatically reconnect: {}

java.io.EOFException: null
    at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.8.0_91]
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:267) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232) ~[activemq-client-5.13.4.jar:5.13.4]
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215) ~[activemq-client-5.13.4.jar:5.13.4]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]

While activeMQ instance logs:

2017-01-05 09:19:11,909 | WARN  | Transport Connection to: tcp://127.0.0.1:57025 failed: java.io.IOException: Frame size of 363 MB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///127.0.0.1:57025@61616
2017-01-05 09:19:11,922 | WARN  | Transport Connection to: tcp://127.0.0.1:57026 failed: java.io.IOException: Frame size of 363 MB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///127.0.0.1:57026@61616

I tried to setup a TransportListener to verify if I can capture this case, but I just receive a transportInterupted event, without any classifier.

I read the documentation about the failover transport (http://activemq.apache.org/failover-transport-reference.html) and maybe I can use the maxReconnectAttempts, but I understand I will have several drawbacks in more common situations (like the server unavailable for a while).

How can I detect this kind of situation and avoid the infinite connection loop between the client and server?

Answer 1

As you said this way

maxReconnectAttempts -1 | 0 From ActiveMQ 5.6: default is -1, retry forever. 0 means disables re-connection, e.g: just try to connect once. Before ActiveMQ 5.6: default is 0, retry forever. All ActiveMQ versions: a value >0 denotes the maximum number of reconnect attempts before an error is sent back to the client.

So if you want your transport listener to be notified of transport failure after failing retries due to the size of your message you need to set maxReconnectAttempts to a value > 0 after that when the max retries reached the method onException of your transport listener will be called with IOException as parameter but as you said it is not easy to verify if it is due to max size or another issue.

If you want to check message size as proposed before sending you can get maxFrameSize configured in the uri on the broker side at runtime by accessing it by jmx and get a BrokerViewMBean instance and call getTransportConnectorByType method http://activemq.apache.org/maven/apidocs/src-html/org/apache/activemq/broker/jmx/BrokerViewMBean.html#line.304 this will return the uri configured in activemq.xml which you can parse to retrieve maxFrameSize.

JMXServiceURL url = new     JMXServiceURL("service:jmx:rmi:///jndi/rmi://hist:1099/jmxrmi");
JMXConnector jmxc = JMXConnectorFactory.connect(url);
MBeanServerConnection conn = jmxc.getMBeanServerConnection(); 

ObjectName activeMq = new ObjectName("org.apache.activemq:Type=Broker,BrokerName=localhost");

BrokerViewMBean mbean = (BrokerViewMBean)MBeanServerInvocationHandler.newProxyInstance(conn, activeMq, BrokerViewMBean.class, true);
String uri = mbean.getTransportConnectorByType("tcp");// or ("ssl") 
String[] pairs = uri.split("&");
for (String pair : pairs) {
    if (pair.startsWith("wireFormat.maxFrameSize")) {
        int idx = pair.indexOf("=");
        System.out.println(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
    }
}

http://activemq.apache.org/maven/apidocs/org/apache/activemq/broker/jmx/BrokerViewMBean.html#getTransportConnectors-- will return a map of transports names as keys and uri's as values

to have a better size of a message you can do that :

        OpenWireFormat opf = new OpenWireFormat();
        opf.setTightEncodingEnabled(true);
        ByteSequence tab = opf.marshal(message);
        System.out.println(tab.length);

your business must be like this :

import java.io.IOException;
import java.net.URLDecoder;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.transport.failover.FailoverTransport;
import org.apache.activemq.util.ByteSequence;

public class SimpleSenderMaxSizeManager {

    private static Connection conn = null;
    private static boolean transportChanged;
    private static Long MAX_FRAME_SIZE;

    public static void main(String[] args) throws JMSException {
        try {
            SimpleSenderMaxSizeManager.updateMaxSize("host1");
            ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
                    "failover:(tcp://host1:5670,tcp://host2:5671)?randomize=false");
            cf.setTransportListener(new TransportListener() {

                @Override
                public void transportResumed() {
                    if (transportChanged) {
                        transportChanged = false;
                        try {
                            SimpleSenderMaxSizeManager.updateMaxSize(null);
                        } catch (Exception e) {
                        }
                    }
                }

                @Override
                public void transportInterupted() {
                    transportChanged = true;
                }

                @Override
                public void onException(IOException error) {
                }

                @Override
                public void onCommand(Object command) {
                }
            });
            conn = cf.createConnection();
            ActiveMQSession session = (ActiveMQSession) conn.createSession(false,
                    ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
            MessageProducer producer = session.createProducer(session.createQueue("TEST"));
            conn.start();
            ActiveMQTextMessage msg = (ActiveMQTextMessage) session.createTextMessage("test");
            OpenWireFormat opf = new OpenWireFormat();
            opf.setTightEncodingEnabled(true);
            ByteSequence tab = opf.marshal(msg);
            System.out.println(tab.length);
            if (tab.length >= MAX_FRAME_SIZE) {
                throw new RuntimeException(tab.length + ">=" + MAX_FRAME_SIZE);
            }
            producer.send(msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (Exception e) {
                }
            }
        }
    }

    protected static void updateMaxSize(String host) throws Exception {
        JMXConnector jmxc = null;
        try {
            String jmxHost = host;
            String scheme = null;
            if (conn == null) {
                scheme = "tcp";
            } else {
                org.apache.activemq.transport.TransportFilter responseCorrelator = (TransportFilter) ((ActiveMQConnection) conn)
                        .getTransport();
                TransportFilter mutexTransport = (TransportFilter) responseCorrelator.getNext();
                FailoverTransport failoverTransport = (FailoverTransport) mutexTransport.getNext();
                while (failoverTransport.getConnectedTransportURI() == null) {
                    try {
                        Thread.sleep(100);
                    } catch (Exception e) {
                    }
                }
                scheme = failoverTransport.getConnectedTransportURI().getScheme();
                jmxHost = failoverTransport.getConnectedTransportURI().getHost();
            }
            JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + jmxHost + ":1099/jmxrmi");
            Map<String, String[]> env = new HashMap<>();
            String[] creds = { "admin", "admin" };
            env.put(JMXConnector.CREDENTIALS, creds);
            jmxc = JMXConnectorFactory.connect(url, env);
            MBeanServerConnection conn = jmxc.getMBeanServerConnection();
            ObjectName activeMq = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost");
            BrokerViewMBean mbean = MBeanServerInvocationHandler.newProxyInstance(conn, activeMq, BrokerViewMBean.class,
                    true);
            String value = mbean.getTransportConnectorByType(scheme);
            String[] pairs = value.split("&");
            for (String pair : pairs) {
                if (pair.contains("wireFormat.maxFrameSize")) {
                    int idx = pair.indexOf("=");
                    System.out.println(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
                    MAX_FRAME_SIZE = Long.valueOf(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
                    MAX_FRAME_SIZE -= 1000;// security for JMS headers added by
                                            // session on sending
                }
            }
        } finally {
            if (jmxc != null) {
                try {
                    jmxc.close();
                } catch (Exception e) {
                }
            }
        }
    }
}