activemq-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Murtaza (JIRA)" <j...@apache.org>
Subject [jira] [Created] (AMQ-4611) Network Failure Issue in Embedded Broker using setStaticBridge=true
Date Mon, 01 Jul 2013 15:00:26 GMT
Murtaza created AMQ-4611:
----------------------------

             Summary: Network Failure Issue in Embedded Broker using setStaticBridge=true
                 Key: AMQ-4611
                 URL: https://issues.apache.org/jira/browse/AMQ-4611
             Project: ActiveMQ
          Issue Type: Bug
          Components: Broker
    Affects Versions: 5.8.0
         Environment: Production
            Reporter: Murtaza
            Priority: Blocker


Hi,

I have an embedded broker connecting to my remote server broker using network connector and
setStaticBridge=true and static list of destinations. When producer produces the message connecting
to my local embedded broker, it successfully forwards to remote broker and stick it to pending
messages. Issue comes during network failure when embedded broker is unable to communicate
to remote broker and producer keep on producing messages. Producer messages will be stored
in KahaDB as a part of persistence setting. When network comes up, I am not able to see old
messages produced by producer in the pending message of the remote broker and it doesn't forward
any new messages produced by the producer to remote broker thereafter. Only temporary solution
is to restart embedded broker. Please help in this urgent issue. My code are as below -

1. Embedded Broker

    public final class EmbeddedBroker {
    public static Destination destination;
    public static String subject = "TOOL.DEFAULT";
    public static boolean topic;
    public static boolean transacted;

    private EmbeddedBroker() {
    }

    public static void main(String[] args) throws Exception {
        BrokerService broker = new BrokerService();
        broker.setUseJmx(true);
        broker.setBrokerName("storeforward");
        broker.addConnector("tcp://localhost:61616");
        broker.setPersistent(true);
        NetworkConnector connector1 = broker.addNetworkConnector("static:failover:"+"tcp://172.16.102.153:61616");
        connector1.setDuplex(true);
        connector1.setStaticBridge(true);

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(null,
null, "failover:tcp://172.16.102.153:61616");
        Connection connection = connectionFactory.createConnection();
        System.out.println("test");
        connection.start();
        Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
        if (topic) {
            destination = session.createTopic(subject);
        } else {
            destination = session.createQueue(subject);
        }
        
        List lst = new ArrayList();
        lst.add(destination);
        connector1.setBridgeTempDestinations(true);
        connector1.setAdvisoryForFailedForward(true);
        connector1.setStaticallyIncludedDestinations(lst);
        
        broker.start();

        // now lets wait forever to avoid the JVM terminating immediately
        Object lock = new Object();
        synchronized (lock) {
            lock.wait();
        }
    }
}


2. Producer

public class ProducerTool extends Thread {

    private Destination destination;
    private int messageCount = 10;
    private long sleepTime;
    private boolean verbose = true;
    private int messageSize = 255;
    private static int parallelThreads = 1;
    private long timeToLive;
/*    private String user = ActiveMQConnection.DEFAULT_USER;
    private String password = ActiveMQConnection.DEFAULT_PASSWORD;
*/    
    private String user = null;
    private String password = null;
    
    private String url = "failover:tcp://172.16.102.153:61616";
    private String subject = "TOOL.DEFAULT";
    private boolean topic;
    private boolean transacted;
    private boolean persistent = true;
    private static Object lockResults = new Object();

    public static void main(String[] args) throws Exception {
    
        ArrayList<ProducerTool> threads = new ArrayList();
        ProducerTool producerTool = new ProducerTool();
        String[] unknown = CommandLineSupport.setOptions(producerTool, args);
        if (unknown.length > 0) {
            System.out.println("Unknown options: " + Arrays.toString(unknown));
            System.exit(-1);
        }
        producerTool.showParameters();
        for (int threadCount = 1; threadCount <= parallelThreads; threadCount++) {
            producerTool = new ProducerTool();
            CommandLineSupport.setOptions(producerTool, args);
            producerTool.start();
            threads.add(producerTool);
        }

        while (true) {
            Iterator<ProducerTool> itr = threads.iterator();
            int running = 0;
            while (itr.hasNext()) {
                ProducerTool thread = itr.next();
                if (thread.isAlive()) {
                    running++;
                }
            }
            if (running <= 0) {
                System.out.println("All threads completed their work");
                break;
            }
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
            }
        }
        System.exit(0);
    }

    public void showParameters() {
        System.out.println("Connecting to URL: " + url + " (" + user + ":" + password + ")");
        System.out.println("Publishing a Message with size " + messageSize + " to " + (topic
? "topic" : "queue") + ": " + subject);
        System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
        System.out.println("Sleeping between publish " + sleepTime + " ms");
        System.out.println("Running " + parallelThreads + " parallel threads");

        if (timeToLive != 0) {
            System.out.println("Messages time to live " + timeToLive + " ms");
        }
    }

    public void run() {
        Connection connection = null;
        try {
            // Create the connection.
            
            url = "tcp://localhost:61616";
            
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,
password, url);
            connection = connectionFactory.createConnection();
            connection.start();
            
            Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
            if (topic) {
                destination = session.createTopic(subject);
            } else {
                destination = session.createQueue(subject);
            }

            // Create the session

            // Create the producer.
            MessageProducer producer = session.createProducer(destination);
            if (persistent) {
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            } else {
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            }
            if (timeToLive != 0) {
                producer.setTimeToLive(timeToLive);
            }

            // Start sending messages
            sendLoop(session, producer);

            System.out.println("[" + this.getName() + "] Done.");

            synchronized (lockResults) {
                ActiveMQConnection c = (ActiveMQConnection) connection;
                System.out.println("[" + this.getName() + "] Results:\n");
                c.getConnectionStats().dump(new IndentPrinter());
            }

        } catch (Exception e) {
            System.out.println("[" + this.getName() + "] Caught: " + e);
            e.printStackTrace();
        } finally {
            try {
                connection.close();
            } catch (Throwable ignore) {
            }
        }
    }

    protected void sendLoop(Session session, MessageProducer producer) throws Exception {

        for (int i = 0; i < messageCount || messageCount == 0; i++) {

            TextMessage message = session.createTextMessage(createMessageText(i));

            if (verbose) {
                String msg = message.getText();
                if (msg.length() > 50) {
                    msg = msg.substring(0, 50) + "...";
                }
                System.out.println("[" + this.getName() + "] Sending message: '" + msg + "'");
            }

            producer.send(message);

            if (transacted) {
                System.out.println("[" + this.getName() + "] Committing " + messageCount +
" messages");
                session.commit();
            }
            Thread.sleep(sleepTime);
        }
    }

    private String createMessageText(int index) {
        StringBuffer buffer = new StringBuffer(messageSize);
        buffer.append("Message: " + index + " sent at: " + new Date());
        if (buffer.length() > messageSize) {
            return buffer.substring(0, messageSize);
        }
        for (int i = buffer.length(); i < messageSize; i++) {
            buffer.append(' ');
        }
        return buffer.toString();
    }

    public void setPersistent(boolean durable) {
        this.persistent = durable;
    }

    public void setMessageCount(int messageCount) {
        this.messageCount = messageCount;
    }

    public void setMessageSize(int messageSize) {
        this.messageSize = messageSize;
    }

    public void setPassword(String pwd) {
        this.password = pwd;
    }

    public void setSleepTime(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    public void setSubject(String subject) {
        this.subject = subject;
    }

    public void setTimeToLive(long timeToLive) {
        this.timeToLive = timeToLive;
    }

    public void setParallelThreads(int parallelThreads) {
        if (parallelThreads < 1) {
            parallelThreads = 1;
        }
        this.parallelThreads = parallelThreads;
    }

    public void setTopic(boolean topic) {
        this.topic = topic;
    }

    public void setQueue(boolean queue) {
        this.topic = !queue;
    }

    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public void setVerbose(boolean verbose) {
        this.verbose = verbose;
    }
}

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

Mime
View raw message