activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r1069813 - /activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java
Date Fri, 11 Feb 2011 14:29:33 GMT
Author: gtully
Date: Fri Feb 11 14:29:33 2011
New Revision: 1069813

URL: http://svn.apache.org/viewvc?rev=1069813&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3167 - add test case from Arthur with thanks.

Added:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java 
 (with props)

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java?rev=1069813&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java Fri
Feb 11 14:29:33 2011
@@ -0,0 +1,509 @@
+/**
+ * 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.
+ */
+
+package org.apache.activemq.bugs;
+
+import java.util.ArrayList;
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Test the loss of messages detected during testing with ActiveMQ 5.4.1 and 5.4.2.
+ * <p/>
+ * Symptoms:
+ * - 1 record is lost "early" in the stream.
+ * - no more records lost.
+ * <p/>
+ * Test Configuration:
+ * - Broker Settings:
+ * - Destination Policy
+ * - Occurs with "Destination Policy" using Store Cursor and a memory limit
+ * - Not reproduced without "Destination Policy" defined
+ * - Persistence Adapter
+ * - Memory: Does not occur.
+ * - KahaDB: Occurs.
+ * - Messages
+ * - Occurs with TextMessage and BinaryMessage
+ * - Persistent messages.
+ * <p/>
+ * Notes:
+ * - Lower memory limits increase the rate of occurrence.
+ * - Higher memory limits may prevent the problem (probably because memory limits not reached).
+ * - Producers sending a number of messages before consumers come online increases rate of
occurrence.
+ */
+
+public class AMQ3167Test {
+    protected BrokerService embeddedBroker;
+
+    protected static final int MEMORY_LIMIT = 16 * 1024;
+
+    protected static boolean Debug_f = false;
+
+    protected long Producer_stop_time = 0;
+    protected long Consumer_stop_time = 0;
+    protected long Consumer_startup_delay_ms = 2000;
+    protected boolean Stop_after_error = true;
+
+    protected Connection JMS_conn;
+    protected long Num_error = 0;
+
+
+    ////             ////
+    ////  UTILITIES  ////
+    ////             ////
+
+
+    /**
+     * Create a new, unsecured, client connection to the test broker using the given username
and password.  This
+     * connection bypasses all security.
+     * <p/>
+     * Don't forget to start the connection or no messages will be received by consumers
even though producers
+     * will work fine.
+     *
+     * @username name of the JMS user for the connection; may be null.
+     * @password Password for the JMS user; may be null.
+     */
+
+    protected Connection createUnsecuredConnection(String username, String password)
+            throws javax.jms.JMSException {
+        ActiveMQConnectionFactory conn_fact;
+
+        conn_fact = new ActiveMQConnectionFactory(embeddedBroker.getVmConnectorURI());
+
+        return conn_fact.createConnection(username, password);
+    }
+
+
+    ////                      ////
+    ////  TEST FUNCTIONALITY  ////
+    ////                      ////
+
+
+    @Before
+    public void testPrep()
+            throws Exception {
+        embeddedBroker = new BrokerService();
+        configureBroker(embeddedBroker);
+        embeddedBroker.start();
+        embeddedBroker.waitUntilStarted();
+
+        // Prepare the connection
+        JMS_conn = createUnsecuredConnection(null, null);
+        JMS_conn.start();
+    }
+
+    @After
+    public void testCleanup()
+            throws java.lang.Exception {
+        JMS_conn.stop();
+        embeddedBroker.stop();
+    }
+
+
+    protected void configureBroker(BrokerService broker_svc)
+            throws Exception {
+        TransportConnector conn;
+
+        broker_svc.setBrokerName("testbroker1");
+
+        broker_svc.setUseJmx(false);
+        broker_svc.setPersistent(true);
+        broker_svc.setDataDirectory("target/AMQ3167Test");
+        configureDestinationPolicy(broker_svc);
+    }
+
+
+    /**
+     * NOTE: overrides any prior policy map defined for the broker service.
+     */
+
+    protected void configureDestinationPolicy(BrokerService broker_svc) {
+        PolicyMap pol_map;
+        PolicyEntry pol_ent;
+        ArrayList<PolicyEntry> ent_list;
+
+        ent_list = new ArrayList<PolicyEntry>();
+
+        //
+        // QUEUES
+        //
+
+        pol_ent = new PolicyEntry();
+        pol_ent.setQueue(">");
+        pol_ent.setMemoryLimit(MEMORY_LIMIT);
+        pol_ent.setProducerFlowControl(false);
+        ent_list.add(pol_ent);
+
+
+        //
+        // COMPLETE POLICY MAP
+        //
+
+        pol_map = new PolicyMap();
+        pol_map.setPolicyEntries(ent_list);
+
+        broker_svc.setDestinationPolicy(pol_map);
+    }
+
+
+    ////        ////
+    ////  TEST  ////
+    ////        ////
+
+    @Test
+    public void testQueueLostMessage()
+            throws Exception {
+        Destination dest;
+
+        dest = ActiveMQDestination.createDestination("lostmsgtest.queue", ActiveMQDestination.QUEUE_TYPE);
+
+        // 10 seconds from now
+        Producer_stop_time = java.lang.System.nanoTime() + (10L * 1000000000L);
+
+        // 15 seconds from now
+        Consumer_stop_time = Producer_stop_time + (5L * 1000000000L);
+
+        runLostMsgTest(dest, 1000000, 1, 1, false);
+
+        // Make sure failures in the threads are thoroughly reported in the JUnit framework.
+        assertTrue(Num_error == 0);
+    }
+
+
+    /**
+     *
+     */
+
+    protected static void log(String msg) {
+        if (Debug_f)
+            java.lang.System.err.println(msg);
+    }
+
+
+    /**
+     * Main body of the lost-message test.
+     */
+
+    protected void runLostMsgTest(Destination dest, int num_msg, int num_send_per_sess, int
num_recv_per_sess,
+                                  boolean topic_f)
+            throws Exception {
+        Thread prod_thread;
+        Thread cons_thread;
+        String tag;
+        Session sess;
+        MessageProducer prod;
+        MessageConsumer cons;
+        int ack_mode;
+
+
+        //
+        // Start the producer
+        //
+
+        tag = "prod";
+        log(">> Starting producer " + tag);
+
+        sess = JMS_conn.createSession((num_send_per_sess > 1), Session.AUTO_ACKNOWLEDGE);
+        prod = sess.createProducer(dest);
+
+        prod_thread = new producerThread(sess, prod, tag, num_msg, num_send_per_sess);
+        prod_thread.start();
+        log("Started producer " + tag);
+
+
+        //
+        // Delay before starting consumers
+        //
+
+        log("Waiting before starting consumers");
+        java.lang.Thread.sleep(Consumer_startup_delay_ms);
+
+
+        //
+        // Now create and start the consumer
+        //
+
+        tag = "cons";
+        log(">> Starting consumer");
+
+        if (num_recv_per_sess > 1)
+            ack_mode = Session.CLIENT_ACKNOWLEDGE;
+        else
+            ack_mode = Session.AUTO_ACKNOWLEDGE;
+
+        sess = JMS_conn.createSession(false, ack_mode);
+        cons = sess.createConsumer(dest);
+
+        cons_thread = new consumerThread(sess, cons, tag, num_msg, num_recv_per_sess);
+        cons_thread.start();
+        log("Started consumer " + tag);
+
+
+        //
+        // Wait for the producer and consumer to finish.
+        //
+
+        log("< waiting for producer.");
+        prod_thread.join();
+
+        log("< waiting for consumer.");
+        cons_thread.join();
+
+        log("Shutting down");
+    }
+
+
+    ////                    ////
+    ////  INTERNAL CLASSES  ////
+    ////                    ////
+
+    /**
+     * Producer thread - runs a single producer until the maximum number of messages is sent,
the producer stop
+     * time is reached, or a test error is detected.
+     */
+
+    protected class producerThread extends Thread {
+        protected Session msgSess;
+        protected MessageProducer msgProd;
+        protected String producerTag;
+        protected int numMsg;
+        protected int numPerSess;
+        protected long producer_stop_time;
+
+        producerThread(Session sess, MessageProducer prod, String tag, int num_msg, int sess_size)
{
+            super();
+
+            producer_stop_time = 0;
+            msgSess = sess;
+            msgProd = prod;
+            producerTag = tag;
+            numMsg = num_msg;
+            numPerSess = sess_size;
+        }
+
+        public void execTest()
+                throws Exception {
+            Message msg;
+            int sess_start;
+            int cur;
+
+            sess_start = 0;
+            cur = 0;
+            while ((cur < numMsg) && (!didTimeOut()) &&
+                    ((!Stop_after_error) || (Num_error == 0))) {
+                msg = msgSess.createTextMessage("test message from " + producerTag);
+                msg.setStringProperty("testprodtag", producerTag);
+                msg.setIntProperty("seq", cur);
+
+                if (msg instanceof ActiveMQMessage) {
+                    ((ActiveMQMessage) msg).setResponseRequired(true);
+                }
+
+
+                //
+                // Send the message.
+                //
+
+                msgProd.send(msg);
+                cur++;
+
+
+                //
+                // Commit if the number of messages per session has been reached, and
+                //  transactions are being used (only when > 1 msg per sess).
+                //
+
+                if ((numPerSess > 1) && ((cur - sess_start) >= numPerSess))
{
+                    msgSess.commit();
+                    sess_start = cur;
+                }
+            }
+
+            // Make sure to send the final commit, if there were sends since the last commit.
+            if ((numPerSess > 1) && ((cur - sess_start) > 0))
+                msgSess.commit();
+
+            if (cur < numMsg)
+                log("* Producer " + producerTag + " timed out at " + java.lang.System.nanoTime()
+
+                        " (stop time " + producer_stop_time + ")");
+        }
+
+
+        /**
+         * Check whether it is time for the producer to terminate.
+         */
+
+        protected boolean didTimeOut() {
+            if ((Producer_stop_time > 0) && (java.lang.System.nanoTime() >=
Producer_stop_time))
+                return true;
+
+            return false;
+        }
+
+        /**
+         * Run the producer.
+         */
+
+        @Override
+        public void run() {
+            try {
+                log("- running producer " + producerTag);
+                execTest();
+                log("- finished running producer " + producerTag);
+            } catch (Throwable thrown) {
+                Num_error++;
+                fail("producer " + producerTag + " failed: " + thrown.getMessage());
+                throw new Error("producer " + producerTag + " failed", thrown);
+            }
+        }
+
+        @Override
+        public String toString() {
+            return producerTag;
+        }
+    }
+
+
+    /**
+     * Producer thread - runs a single consumer until the maximum number of messages is received,
the consumer stop
+     * time is reached, or a test error is detected.
+     */
+
+    protected class consumerThread extends Thread {
+        protected Session msgSess;
+        protected MessageConsumer msgCons;
+        protected String consumerTag;
+        protected int numMsg;
+        protected int numPerSess;
+
+        consumerThread(Session sess, MessageConsumer cons, String tag, int num_msg, int sess_size)
{
+            super();
+
+            msgSess = sess;
+            msgCons = cons;
+            consumerTag = tag;
+            numMsg = num_msg;
+            numPerSess = sess_size;
+        }
+
+        public void execTest()
+                throws Exception {
+            Message msg;
+            int sess_start;
+            int cur;
+
+            msg = null;
+            sess_start = 0;
+            cur = 0;
+
+            while ((cur < numMsg) && (!didTimeOut()) &&
+                    ((!Stop_after_error) || (Num_error == 0))) {
+                //
+                // Use a timeout of 1 second to periodically check the consumer timeout.
+                //
+                msg = msgCons.receive(1000);
+                if (msg != null) {
+                    checkMessage(msg, cur);
+                    cur++;
+
+                    if ((numPerSess > 1) && ((cur - sess_start) >= numPerSess))
{
+                        msg.acknowledge();
+                        sess_start = cur;
+                    }
+                }
+            }
+
+            // Acknowledge the last messages, if they were not yet acknowledged.
+            if ((numPerSess > 1) && ((cur - sess_start) > 0))
+                msg.acknowledge();
+
+            if (cur < numMsg)
+                log("* Consumer " + consumerTag + " timed out");
+        }
+
+
+        /**
+         * Check whether it is time for the consumer to terminate.
+         */
+
+        protected boolean didTimeOut() {
+            if ((Consumer_stop_time > 0) && (java.lang.System.nanoTime() >=
Consumer_stop_time))
+                return true;
+
+            return false;
+        }
+
+
+        /**
+         * Verify the message received.  Sequence numbers are checked and are expected to
exactly match the
+         * message number (starting at 0).
+         */
+
+        protected void checkMessage(Message msg, int exp_seq)
+                throws javax.jms.JMSException {
+            int seq;
+
+            seq = msg.getIntProperty("seq");
+
+            if (exp_seq != seq) {
+                Num_error++;
+                fail("*** Consumer " + consumerTag + " expected seq " + exp_seq + "; received
" + seq);
+            }
+        }
+
+
+        /**
+         * Run the consumer.
+         */
+
+        @Override
+        public void run() {
+            try {
+                log("- running consumer " + consumerTag);
+                execTest();
+                log("- running consumer " + consumerTag);
+            } catch (Throwable thrown) {
+                Num_error++;
+                fail("consumer " + consumerTag + " failed: " + thrown.getMessage());
+                throw new Error("consumer " + consumerTag + " failed", thrown);
+            }
+        }
+
+        @Override
+        public String toString() {
+            return consumerTag;
+        }
+    }
+}
\ No newline at end of file

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3167Test.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date



Mime
View raw message