bdbstore/systests/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBUpgradeTest.java - qpid-broker-j - Git at Google

 /*
  *
  * 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.qpid.server.store.berkeleydb;

 import static org.apache.qpid.systests.JmsTestBase.DEFAULT_BROKER_CONFIG;
 import static org.apache.qpid.systests.Utils.INDEX;
 import static org.apache.qpid.systests.Utils.sendMessages;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeThat;

 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;

 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Queue;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 import javax.jms.Topic;
 import javax.jms.TopicConnection;
 import javax.jms.TopicPublisher;
 import javax.jms.TopicSession;
 import javax.jms.TopicSubscriber;

 import org.junit.BeforeClass;
 import org.junit.Test;

 import org.apache.qpid.server.model.AlternateBinding;
 import org.apache.qpid.server.model.Exchange;
 import org.apache.qpid.server.model.ExclusivityPolicy;
 import org.apache.qpid.server.model.Protocol;
 import org.apache.qpid.systests.AmqpManagementFacade;
 import org.apache.qpid.tests.utils.ConfigItem;

 /**
  * Tests upgrading a BDB store on broker startup.
  * The store will then be used to verify that the upgrade is completed
  * properly and that once upgraded it functions as expected.
  * <p>
  * Store prepared using old client/broker with BDBStoreUpgradeTestPreparer.
  */
 @ConfigItem(name = "qpid.initialConfigurationLocation", value = DEFAULT_BROKER_CONFIG )
 public class BDBUpgradeTest extends UpgradeTestBase
 {

     private static final String STRING_1024 = generateString(1024);
     private static final String STRING_1024_256 = generateString(1024 * 256);

     private static final String TOPIC_NAME = "myUpgradeTopic";
     private static final String SUB_NAME = "myDurSubName";
     private static final String SELECTOR_SUB_NAME = "mySelectorDurSubName";
     private static final String SELECTOR_TOPIC_NAME = "mySelectorUpgradeTopic";
     private static final String QUEUE_NAME = "myUpgradeQueue";
     private static final String PRIORITY_QUEUE_NAME = "myPriorityQueue";
     private static final String QUEUE_WITH_DLQ_NAME = "myQueueWithDLQ";

     @BeforeClass
     public static void verifyClient()
     {
         assumeThat(System.getProperty("virtualhostnode.type", "BDB"), is(equalTo("BDB")));
         assumeThat(getProtocol(), is(not(equalTo(Protocol.AMQP_1_0))));
     }

     /**
      * Test that the selector applied to the DurableSubscription was successfully
      * transferred to the new store, and functions as expected with continued use
      * by monitoring message count while sending new messages to the topic and then
      * consuming them.
      */
     @Test
     public void testSelectorDurability() throws Exception
     {
         TopicConnection connection = getTopicConnection();
         try
         {
             connection.start();

             TopicSession session = connection.createTopicSession(true, Session.SESSION_TRANSACTED);
             Topic topic = session.createTopic(SELECTOR_TOPIC_NAME);
             TopicPublisher publisher = session.createPublisher(topic);

             int index = ThreadLocalRandom.current().nextInt();
             Message messageA = session.createTextMessage("A");
             messageA.setIntProperty("ID", index);
             messageA.setStringProperty("testprop", "false");
             publisher.publish(messageA);

             Message messageB = session.createTextMessage("B");
             messageB.setIntProperty("ID", index);
             messageB.setStringProperty("testprop", "true");
             publisher.publish(messageB);

             session.commit();

             TopicSubscriber subscriber =
                     session.createDurableSubscriber(topic, SELECTOR_SUB_NAME, "testprop='true'", false);
             Message migrated = subscriber.receive(getReceiveTimeout());
             assertThat("Failed to receive migrated message", migrated, is(notNullValue()));

             Message received = subscriber.receive(getReceiveTimeout());
             session.commit();
             assertThat("Failed to receive published message", received, is(notNullValue()));
             assertThat("Message is not Text message", received, is(instanceOf(TextMessage.class)));
             assertThat("Unexpected text", ((TextMessage) received).getText(), is(equalTo("B")));
             assertThat("Unexpected index", received.getIntProperty("ID"), is(equalTo(index)));

             session.close();
         }
         finally
         {
             connection.close();
         }
     }

     /**
      * Test that the DurableSubscription without selector was successfully
      * transfered to the new store, and functions as expected with continued use.
      */
     @Test
     public void testDurableSubscriptionWithoutSelector() throws Exception
     {
         TopicConnection connection = getTopicConnection();
         try
         {
             connection.start();

             TopicSession session = connection.createTopicSession(true, Session.SESSION_TRANSACTED);

             Topic topic = session.createTopic(TOPIC_NAME);
             TopicPublisher publisher = session.createPublisher(topic);

             int index = ThreadLocalRandom.current().nextInt();
             Message messageA = session.createTextMessage("A");
             messageA.setIntProperty("ID", index);
             messageA.setStringProperty("testprop", "false");
             publisher.publish(messageA);

             Message messageB = session.createTextMessage("B");
             messageB.setIntProperty("ID", index);
             messageB.setStringProperty("testprop", "true");
             publisher.publish(messageB);

             session.commit();

             TopicSubscriber subscriber = session.createDurableSubscriber(topic, SUB_NAME);
             Message migrated = subscriber.receive(getReceiveTimeout());
             assertThat("Failed to receive migrated message", migrated, is(notNullValue()));

             Message receivedA = subscriber.receive(getReceiveTimeout());
             session.commit();
             assertThat("Failed to receive published message A", receivedA, is(notNullValue()));
             assertThat("Message A is not Text message", receivedA, is(instanceOf(TextMessage.class)));
             assertThat("Unexpected text for A", ((TextMessage) receivedA).getText(), is(equalTo("A")));
             assertThat("Unexpected index", receivedA.getIntProperty("ID"), is(equalTo(index)));

             Message receivedB = subscriber.receive(getReceiveTimeout());
             session.commit();
             assertThat("Failed to receive published message B", receivedB, is(notNullValue()));
             assertThat("Message B is not Text message", receivedB, is(instanceOf(TextMessage.class)));
             assertThat("Unexpected text for B", ((TextMessage) receivedB).getText(), is(equalTo("B")));
             assertThat("Unexpected index  for B", receivedB.getIntProperty("ID"), is(equalTo(index)));

             session.commit();
             session.close();
         }
         finally
         {
             connection.close();
         }
     }

     /**
      * Test that the backing queue for the durable subscription created was successfully
      * detected and set as being exclusive during the upgrade process, and that the
      * regular queue was not.
      */
     @Test
     public void testQueueExclusivity() throws Exception
     {
         Map<String, Object> result = getQueueAttributes(QUEUE_NAME);
         assertThat("Exclusive policy attribute is not found",
                    result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE),
                    is(notNullValue()));
         assertThat("Queue should not have been marked as Exclusive during upgrade",
                    ExclusivityPolicy.valueOf(String.valueOf(result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE))),
                    is(equalTo(ExclusivityPolicy.NONE)));

         result = getQueueAttributes("clientid" + ":" + SUB_NAME);
         assertThat("Exclusive policy attribute is not found",
                    result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE),
                    is(notNullValue()));
         assertThat("DurableSubscription backing queue should have been marked as Exclusive during upgrade",
                    ExclusivityPolicy.valueOf(String.valueOf(result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE))),
                    is(not(equalTo(ExclusivityPolicy.NONE))));
     }

     /**
      * Test that the upgraded queue continues to function properly when used
      * for persistent messaging and restarting the broker.
      * <p>
      * Sends the new messages to the queue BEFORE consuming those which were
      * sent before the upgrade. In doing so, this also serves to test that
      * the queue bindings were successfully transitioned during the upgrade.
      */
     @Test
     public void testBindingAndMessageDurability() throws Exception
     {
         Connection connection = getConnection();
         try
         {
             connection.start();

             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             Queue queue = session.createQueue(QUEUE_NAME);

             sendMessages(connection, queue, 1);

             session.close();

             // Restart
             getBrokerAdmin().restart();

             // Drain the queue of all messages
             connection = getConnection();
             connection.start();
             consumeQueueMessages(connection, true);
         }
         finally
         {
             connection.close();
         }
     }

     /**
      * Test that all of the committed persistent messages previously sent to
      * the broker are properly received following update of the MetaData and
      * Content entries during the store upgrade process.
      */
     @Test
     public void testConsumptionOfUpgradedMessages() throws Exception
     {
         // Create a connection and start it
         Connection connection = getConnection();
         try
         {
             connection.start();

             consumeDurableSubscriptionMessages(connection, true);
             consumeDurableSubscriptionMessages(connection, false);
             consumeQueueMessages(connection, false);
         }
         finally
         {
             connection.close();
         }
     }

     /**
      * Tests store upgrade has maintained the priority queue configuration,
      * such that sending messages with priorities out-of-order and then consuming
      * them gets the messages back in priority order.
      */
     @Test
     public void testPriorityQueue() throws Exception
     {
         Connection connection = getConnection();
         try
         {
             connection.start();

             // send some messages to the priority queue
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             Queue queue = session.createQueue(PRIORITY_QUEUE_NAME);
             MessageProducer producer = session.createProducer(queue);

             producer.send(session.createTextMessage("A"), DeliveryMode.PERSISTENT, 4, -1);
             producer.send(session.createTextMessage("B"), DeliveryMode.PERSISTENT, 1, -1);
             producer.send(session.createTextMessage("C"), DeliveryMode.PERSISTENT, 9, -1);
             session.close();

             //consume the messages, expected order: C, A, B.
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageConsumer consumer = session.createConsumer(queue);

             Message message1 = consumer.receive(getReceiveTimeout());
             assertThat("expected message was not received", message1, is(instanceOf(TextMessage.class)));
             assertThat(((TextMessage) message1).getText(), is(equalTo("C")));
             Message message2 = consumer.receive(getReceiveTimeout());
             assertThat("expected message was not received", message2, is(instanceOf(TextMessage.class)));
             assertThat(((TextMessage) message2).getText(), is(equalTo("A")));
             Message message3 = consumer.receive(getReceiveTimeout());
             assertThat("expected message was not received", message3, is(instanceOf(TextMessage.class)));
             assertThat(((TextMessage) message3).getText(), is(equalTo("B")));
         }
         finally
         {
             connection.close();
         }
     }

     /**
      * Test that the queue configured to have a DLQ was recovered and has the alternate exchange
      * and max delivery count, the DLE exists, the DLQ exists with no max delivery count, the
      * DLQ is bound to the DLE, and that the DLQ does not itself have a DLQ.
      * <p>
      * DLQs are NOT enabled at the virtualhost level, we are testing recovery of the arguments
      * that turned it on for this specific queue.
      */
     @Test
     public void testRecoveryOfQueueWithDLQ() throws Exception
     {
         //verify the DLE exchange exists, has the expected type, and a single binding for the DLQ
         Map<String, Object> exchangeAttributes = getExchangeAttributes(QUEUE_WITH_DLQ_NAME + "_DLE");
         assertThat("Wrong exchange type",
                    exchangeAttributes.get(Exchange.TYPE),
                    is(equalTo("org.apache.qpid.FanoutExchange")));

         @SuppressWarnings("unchecked")
         Collection<Map<String, Object>> bindings = (Collection<Map<String, Object>>) exchangeAttributes.get("bindings");
         assertThat(bindings.size(), is(equalTo(1)));
         for (Map<String, Object> binding : bindings)
         {
             String bindingKey = (String) binding.get("bindingKey");
             String queueName = (String) binding.get("destination");

             //Because its a fanout exchange, we just return a single '*' key with all bound queues
             assertThat("unexpected binding key", bindingKey, is(equalTo("dlq")));
             assertThat("unexpected queue name", queueName, is(equalTo(QUEUE_WITH_DLQ_NAME + "_DLQ")));
         }

         //verify the queue exists, has the expected alternate exchange and max delivery count
         Map<String, Object> queueAttributes = getQueueAttributes(QUEUE_WITH_DLQ_NAME);
         assertThat("Queue does not have the expected AlternateExchange",
                    queueAttributes.get(Exchange.ALTERNATE_BINDING),
                    is(equalTo(Collections.singletonMap(AlternateBinding.DESTINATION, QUEUE_WITH_DLQ_NAME + "_DLE"))));

         assertThat("Unexpected maximum delivery count",
                    ((Number) queueAttributes.get(org.apache.qpid.server.model.Queue.MAXIMUM_DELIVERY_ATTEMPTS)).intValue(),
                    is(equalTo(2)));

         Map<String, Object> dlQueueAttributes = getQueueAttributes(QUEUE_WITH_DLQ_NAME + "_DLQ");
         assertThat("Queue should not have an AlternateExchange",
                    dlQueueAttributes.get(org.apache.qpid.server.model.Queue.ALTERNATE_BINDING), is(nullValue()));
         assertThat("Unexpected maximum delivery count",
                    ((Number) dlQueueAttributes.get(org.apache.qpid.server.model.Queue.MAXIMUM_DELIVERY_ATTEMPTS)).intValue(),
                    is(equalTo(0)));

         try
         {
             String queueName = QUEUE_WITH_DLQ_NAME + "_DLQ_DLQ";
             getQueueAttributes(queueName);
             fail("A DLQ should not exist for the DLQ itself");
         }
         catch (AmqpManagementFacade.OperationUnsuccessfulException e)
         {
             assertThat(e.getStatusCode(), is(equalTo(404)));
         }
     }

     @Override
     String getOldStoreResourcePath()
     {
         return "upgrade/bdbstore-v4/test-store/00000000.jdb";
     }

     private Map<String, Object> getExchangeAttributes(final String exchangeName) throws Exception
     {
         return readEntityUsingAmqpManagement(exchangeName, "org.apache.qpid.Exchange", false);
     }

     private Map<String, Object> getQueueAttributes(final String queueName) throws Exception
     {
         return readEntityUsingAmqpManagement(queueName, "org.apache.qpid.Queue", false);
     }

     private void consumeDurableSubscriptionMessages(Connection connection, boolean selector) throws Exception
     {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Topic topic = null;
         TopicSubscriber durSub = null;

         if (selector)
         {
             topic = session.createTopic(SELECTOR_TOPIC_NAME);
             durSub = session.createDurableSubscriber(topic, SELECTOR_SUB_NAME, "testprop='true'", false);
         }
         else
         {
             topic = session.createTopic(TOPIC_NAME);
             durSub = session.createDurableSubscriber(topic, SUB_NAME);
         }

         // Retrieve the matching message
         Message m = durSub.receive(getReceiveTimeout());
         assertThat("Failed to receive an expected message", m, is(notNullValue()));
         if (selector)
         {
             assertThat("Selector property did not match", m.getStringProperty("testprop"), is(equalTo("true")));
         }
         assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(1)));
         assertThat("Message content was not as expected",
                    ((TextMessage) m).getText(),
                    is(equalTo(generateString(1024))));

         // Verify that no more messages are received
         m = durSub.receive(getReceiveTimeout());
         assertThat("No more messages should have been recieved", m, is(nullValue()));

         durSub.close();
         session.close();
     }

     private void consumeQueueMessages(Connection connection, boolean extraMessage) throws Exception
     {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Queue queue = session.createQueue(QUEUE_NAME);

         MessageConsumer consumer = session.createConsumer(queue);
         Message m;

         // Retrieve the initial pre-upgrade messages
         for (int i = 1; i <= 5; i++)
         {
             m = consumer.receive(getReceiveTimeout());
             assertThat("Failed to receive an expected message", m, is(notNullValue()));
             assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(i)));
             assertThat("Message content was not as expected",
                        ((TextMessage) m).getText(),
                        is(equalTo(STRING_1024_256)));
         }
         for (int i = 1; i <= 5; i++)
         {
             m = consumer.receive(getReceiveTimeout());
             assertThat("Failed to receive an expected message", m, is(notNullValue()));
             assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(i)));
             assertThat("Message content was not as expected", ((TextMessage) m).getText(), is((equalTo(STRING_1024))));
         }

         if (extraMessage)
         {
             //verify that the extra message is received
             m = consumer.receive(getReceiveTimeout());
             assertThat("Failed to receive an expected message", m, is(notNullValue()));
             assertThat("ID property did not match", m.getIntProperty(INDEX), is(equalTo(0)));
         }
         else
         {
             // Verify that no more messages are received
             m = consumer.receive(getReceiveTimeout());
             assertThat("No more messages should have been recieved", m, is(nullValue()));
         }

         consumer.close();
         session.close();
     }

     /**
      * Generates a string of a given length consisting of the sequence 0,1,2,..,9,0,1,2.
      *
      * @param length number of characters in the string
      * @return string sequence of the given length
      */
     private static String generateString(int length)
     {
         char[] base_chars = new char[]{'0','1','2','3','4','5','6','7','8','9'};
         char[] chars = new char[length];
         for (int i = 0; i < (length); i++)
         {
             chars[i] = base_chars[i % 10];
         }
         return new String(chars);
     }

 }
	/*
	*
	* 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.qpid.server.store.berkeleydb;

	import static org.apache.qpid.systests.JmsTestBase.DEFAULT_BROKER_CONFIG;
	import static org.apache.qpid.systests.Utils.INDEX;
	import static org.apache.qpid.systests.Utils.sendMessages;
	import static org.hamcrest.CoreMatchers.equalTo;
	import static org.hamcrest.CoreMatchers.instanceOf;
	import static org.hamcrest.CoreMatchers.is;
	import static org.hamcrest.CoreMatchers.not;
	import static org.hamcrest.CoreMatchers.notNullValue;
	import static org.hamcrest.CoreMatchers.nullValue;
	import static org.hamcrest.MatcherAssert.assertThat;
	import static org.junit.Assert.fail;
	import static org.junit.Assume.assumeThat;

	import java.util.Collection;
	import java.util.Collections;
	import java.util.Map;
	import java.util.concurrent.ThreadLocalRandom;

	import javax.jms.Connection;
	import javax.jms.DeliveryMode;
	import javax.jms.Message;
	import javax.jms.MessageConsumer;
	import javax.jms.MessageProducer;
	import javax.jms.Queue;
	import javax.jms.Session;
	import javax.jms.TextMessage;
	import javax.jms.Topic;
	import javax.jms.TopicConnection;
	import javax.jms.TopicPublisher;
	import javax.jms.TopicSession;
	import javax.jms.TopicSubscriber;

	import org.junit.BeforeClass;
	import org.junit.Test;

	import org.apache.qpid.server.model.AlternateBinding;
	import org.apache.qpid.server.model.Exchange;
	import org.apache.qpid.server.model.ExclusivityPolicy;
	import org.apache.qpid.server.model.Protocol;
	import org.apache.qpid.systests.AmqpManagementFacade;
	import org.apache.qpid.tests.utils.ConfigItem;

	/**
	* Tests upgrading a BDB store on broker startup.
	* The store will then be used to verify that the upgrade is completed
	* properly and that once upgraded it functions as expected.
	* <p>
	* Store prepared using old client/broker with BDBStoreUpgradeTestPreparer.
	*/
	@ConfigItem(name = "qpid.initialConfigurationLocation", value = DEFAULT_BROKER_CONFIG )
	public class BDBUpgradeTest extends UpgradeTestBase
	{

	private static final String STRING_1024 = generateString(1024);
	private static final String STRING_1024_256 = generateString(1024 * 256);

	private static final String TOPIC_NAME = "myUpgradeTopic";
	private static final String SUB_NAME = "myDurSubName";
	private static final String SELECTOR_SUB_NAME = "mySelectorDurSubName";
	private static final String SELECTOR_TOPIC_NAME = "mySelectorUpgradeTopic";
	private static final String QUEUE_NAME = "myUpgradeQueue";
	private static final String PRIORITY_QUEUE_NAME = "myPriorityQueue";
	private static final String QUEUE_WITH_DLQ_NAME = "myQueueWithDLQ";

	@BeforeClass
	public static void verifyClient()
	{
	assumeThat(System.getProperty("virtualhostnode.type", "BDB"), is(equalTo("BDB")));
	assumeThat(getProtocol(), is(not(equalTo(Protocol.AMQP_1_0))));
	}

	/**
	* Test that the selector applied to the DurableSubscription was successfully
	* transferred to the new store, and functions as expected with continued use
	* by monitoring message count while sending new messages to the topic and then
	* consuming them.
	*/
	@Test
	public void testSelectorDurability() throws Exception
	{
	TopicConnection connection = getTopicConnection();
	try
	{
	connection.start();

	TopicSession session = connection.createTopicSession(true, Session.SESSION_TRANSACTED);
	Topic topic = session.createTopic(SELECTOR_TOPIC_NAME);
	TopicPublisher publisher = session.createPublisher(topic);

	int index = ThreadLocalRandom.current().nextInt();
	Message messageA = session.createTextMessage("A");
	messageA.setIntProperty("ID", index);
	messageA.setStringProperty("testprop", "false");
	publisher.publish(messageA);

	Message messageB = session.createTextMessage("B");
	messageB.setIntProperty("ID", index);
	messageB.setStringProperty("testprop", "true");
	publisher.publish(messageB);

	session.commit();

	TopicSubscriber subscriber =
	session.createDurableSubscriber(topic, SELECTOR_SUB_NAME, "testprop='true'", false);
	Message migrated = subscriber.receive(getReceiveTimeout());
	assertThat("Failed to receive migrated message", migrated, is(notNullValue()));

	Message received = subscriber.receive(getReceiveTimeout());
	session.commit();
	assertThat("Failed to receive published message", received, is(notNullValue()));
	assertThat("Message is not Text message", received, is(instanceOf(TextMessage.class)));
	assertThat("Unexpected text", ((TextMessage) received).getText(), is(equalTo("B")));
	assertThat("Unexpected index", received.getIntProperty("ID"), is(equalTo(index)));

	session.close();
	}
	finally
	{
	connection.close();
	}
	}

	/**
	* Test that the DurableSubscription without selector was successfully
	* transfered to the new store, and functions as expected with continued use.
	*/
	@Test
	public void testDurableSubscriptionWithoutSelector() throws Exception
	{
	TopicConnection connection = getTopicConnection();
	try
	{
	connection.start();

	TopicSession session = connection.createTopicSession(true, Session.SESSION_TRANSACTED);

	Topic topic = session.createTopic(TOPIC_NAME);
	TopicPublisher publisher = session.createPublisher(topic);

	int index = ThreadLocalRandom.current().nextInt();
	Message messageA = session.createTextMessage("A");
	messageA.setIntProperty("ID", index);
	messageA.setStringProperty("testprop", "false");
	publisher.publish(messageA);

	Message messageB = session.createTextMessage("B");
	messageB.setIntProperty("ID", index);
	messageB.setStringProperty("testprop", "true");
	publisher.publish(messageB);

	session.commit();

	TopicSubscriber subscriber = session.createDurableSubscriber(topic, SUB_NAME);
	Message migrated = subscriber.receive(getReceiveTimeout());
	assertThat("Failed to receive migrated message", migrated, is(notNullValue()));

	Message receivedA = subscriber.receive(getReceiveTimeout());
	session.commit();
	assertThat("Failed to receive published message A", receivedA, is(notNullValue()));
	assertThat("Message A is not Text message", receivedA, is(instanceOf(TextMessage.class)));
	assertThat("Unexpected text for A", ((TextMessage) receivedA).getText(), is(equalTo("A")));
	assertThat("Unexpected index", receivedA.getIntProperty("ID"), is(equalTo(index)));

	Message receivedB = subscriber.receive(getReceiveTimeout());
	session.commit();
	assertThat("Failed to receive published message B", receivedB, is(notNullValue()));
	assertThat("Message B is not Text message", receivedB, is(instanceOf(TextMessage.class)));
	assertThat("Unexpected text for B", ((TextMessage) receivedB).getText(), is(equalTo("B")));
	assertThat("Unexpected index for B", receivedB.getIntProperty("ID"), is(equalTo(index)));

	session.commit();
	session.close();
	}
	finally
	{
	connection.close();
	}
	}

	/**
	* Test that the backing queue for the durable subscription created was successfully
	* detected and set as being exclusive during the upgrade process, and that the
	* regular queue was not.
	*/
	@Test
	public void testQueueExclusivity() throws Exception
	{
	Map<String, Object> result = getQueueAttributes(QUEUE_NAME);
	assertThat("Exclusive policy attribute is not found",
	result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE),
	is(notNullValue()));
	assertThat("Queue should not have been marked as Exclusive during upgrade",
	ExclusivityPolicy.valueOf(String.valueOf(result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE))),
	is(equalTo(ExclusivityPolicy.NONE)));

	result = getQueueAttributes("clientid" + ":" + SUB_NAME);
	assertThat("Exclusive policy attribute is not found",
	result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE),
	is(notNullValue()));
	assertThat("DurableSubscription backing queue should have been marked as Exclusive during upgrade",
	ExclusivityPolicy.valueOf(String.valueOf(result.get(org.apache.qpid.server.model.Queue.EXCLUSIVE))),
	is(not(equalTo(ExclusivityPolicy.NONE))));
	}

	/**
	* Test that the upgraded queue continues to function properly when used
	* for persistent messaging and restarting the broker.
	* <p>
	* Sends the new messages to the queue BEFORE consuming those which were
	* sent before the upgrade. In doing so, this also serves to test that
	* the queue bindings were successfully transitioned during the upgrade.
	*/
	@Test
	public void testBindingAndMessageDurability() throws Exception
	{
	Connection connection = getConnection();
	try
	{
	connection.start();

	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	Queue queue = session.createQueue(QUEUE_NAME);

	sendMessages(connection, queue, 1);

	session.close();

	// Restart
	getBrokerAdmin().restart();

	// Drain the queue of all messages
	connection = getConnection();
	connection.start();
	consumeQueueMessages(connection, true);
	}
	finally
	{
	connection.close();
	}
	}

	/**
	* Test that all of the committed persistent messages previously sent to
	* the broker are properly received following update of the MetaData and
	* Content entries during the store upgrade process.
	*/
	@Test
	public void testConsumptionOfUpgradedMessages() throws Exception
	{
	// Create a connection and start it
	Connection connection = getConnection();
	try
	{
	connection.start();

	consumeDurableSubscriptionMessages(connection, true);
	consumeDurableSubscriptionMessages(connection, false);
	consumeQueueMessages(connection, false);
	}
	finally
	{
	connection.close();
	}
	}

	/**
	* Tests store upgrade has maintained the priority queue configuration,
	* such that sending messages with priorities out-of-order and then consuming
	* them gets the messages back in priority order.
	*/
	@Test
	public void testPriorityQueue() throws Exception
	{
	Connection connection = getConnection();
	try
	{
	connection.start();

	// send some messages to the priority queue
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	Queue queue = session.createQueue(PRIORITY_QUEUE_NAME);
	MessageProducer producer = session.createProducer(queue);

	producer.send(session.createTextMessage("A"), DeliveryMode.PERSISTENT, 4, -1);
	producer.send(session.createTextMessage("B"), DeliveryMode.PERSISTENT, 1, -1);
	producer.send(session.createTextMessage("C"), DeliveryMode.PERSISTENT, 9, -1);
	session.close();

	//consume the messages, expected order: C, A, B.
	session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageConsumer consumer = session.createConsumer(queue);

	Message message1 = consumer.receive(getReceiveTimeout());
	assertThat("expected message was not received", message1, is(instanceOf(TextMessage.class)));
	assertThat(((TextMessage) message1).getText(), is(equalTo("C")));
	Message message2 = consumer.receive(getReceiveTimeout());
	assertThat("expected message was not received", message2, is(instanceOf(TextMessage.class)));
	assertThat(((TextMessage) message2).getText(), is(equalTo("A")));
	Message message3 = consumer.receive(getReceiveTimeout());
	assertThat("expected message was not received", message3, is(instanceOf(TextMessage.class)));
	assertThat(((TextMessage) message3).getText(), is(equalTo("B")));
	}
	finally
	{
	connection.close();
	}
	}

	/**
	* Test that the queue configured to have a DLQ was recovered and has the alternate exchange
	* and max delivery count, the DLE exists, the DLQ exists with no max delivery count, the
	* DLQ is bound to the DLE, and that the DLQ does not itself have a DLQ.
	* <p>
	* DLQs are NOT enabled at the virtualhost level, we are testing recovery of the arguments
	* that turned it on for this specific queue.
	*/
	@Test
	public void testRecoveryOfQueueWithDLQ() throws Exception
	{
	//verify the DLE exchange exists, has the expected type, and a single binding for the DLQ
	Map<String, Object> exchangeAttributes = getExchangeAttributes(QUEUE_WITH_DLQ_NAME + "_DLE");
	assertThat("Wrong exchange type",
	exchangeAttributes.get(Exchange.TYPE),
	is(equalTo("org.apache.qpid.FanoutExchange")));

	@SuppressWarnings("unchecked")
	Collection<Map<String, Object>> bindings = (Collection<Map<String, Object>>) exchangeAttributes.get("bindings");
	assertThat(bindings.size(), is(equalTo(1)));
	for (Map<String, Object> binding : bindings)
	{
	String bindingKey = (String) binding.get("bindingKey");
	String queueName = (String) binding.get("destination");

	//Because its a fanout exchange, we just return a single '*' key with all bound queues
	assertThat("unexpected binding key", bindingKey, is(equalTo("dlq")));
	assertThat("unexpected queue name", queueName, is(equalTo(QUEUE_WITH_DLQ_NAME + "_DLQ")));
	}

	//verify the queue exists, has the expected alternate exchange and max delivery count
	Map<String, Object> queueAttributes = getQueueAttributes(QUEUE_WITH_DLQ_NAME);
	assertThat("Queue does not have the expected AlternateExchange",
	queueAttributes.get(Exchange.ALTERNATE_BINDING),
	is(equalTo(Collections.singletonMap(AlternateBinding.DESTINATION, QUEUE_WITH_DLQ_NAME + "_DLE"))));

	assertThat("Unexpected maximum delivery count",
	((Number) queueAttributes.get(org.apache.qpid.server.model.Queue.MAXIMUM_DELIVERY_ATTEMPTS)).intValue(),
	is(equalTo(2)));

	Map<String, Object> dlQueueAttributes = getQueueAttributes(QUEUE_WITH_DLQ_NAME + "_DLQ");
	assertThat("Queue should not have an AlternateExchange",
	dlQueueAttributes.get(org.apache.qpid.server.model.Queue.ALTERNATE_BINDING), is(nullValue()));
	assertThat("Unexpected maximum delivery count",
	((Number) dlQueueAttributes.get(org.apache.qpid.server.model.Queue.MAXIMUM_DELIVERY_ATTEMPTS)).intValue(),
	is(equalTo(0)));

	try
	{
	String queueName = QUEUE_WITH_DLQ_NAME + "_DLQ_DLQ";
	getQueueAttributes(queueName);
	fail("A DLQ should not exist for the DLQ itself");
	}
	catch (AmqpManagementFacade.OperationUnsuccessfulException e)
	{
	assertThat(e.getStatusCode(), is(equalTo(404)));
	}
	}

	@Override
	String getOldStoreResourcePath()
	{
	return "upgrade/bdbstore-v4/test-store/00000000.jdb";
	}

	private Map<String, Object> getExchangeAttributes(final String exchangeName) throws Exception
	{
	return readEntityUsingAmqpManagement(exchangeName, "org.apache.qpid.Exchange", false);
	}

	private Map<String, Object> getQueueAttributes(final String queueName) throws Exception
	{
	return readEntityUsingAmqpManagement(queueName, "org.apache.qpid.Queue", false);
	}

	private void consumeDurableSubscriptionMessages(Connection connection, boolean selector) throws Exception
	{
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	Topic topic = null;
	TopicSubscriber durSub = null;

	if (selector)
	{
	topic = session.createTopic(SELECTOR_TOPIC_NAME);
	durSub = session.createDurableSubscriber(topic, SELECTOR_SUB_NAME, "testprop='true'", false);
	}
	else
	{
	topic = session.createTopic(TOPIC_NAME);
	durSub = session.createDurableSubscriber(topic, SUB_NAME);
	}

	// Retrieve the matching message
	Message m = durSub.receive(getReceiveTimeout());
	assertThat("Failed to receive an expected message", m, is(notNullValue()));
	if (selector)
	{
	assertThat("Selector property did not match", m.getStringProperty("testprop"), is(equalTo("true")));
	}
	assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(1)));
	assertThat("Message content was not as expected",
	((TextMessage) m).getText(),
	is(equalTo(generateString(1024))));

	// Verify that no more messages are received
	m = durSub.receive(getReceiveTimeout());
	assertThat("No more messages should have been recieved", m, is(nullValue()));

	durSub.close();
	session.close();
	}

	private void consumeQueueMessages(Connection connection, boolean extraMessage) throws Exception
	{
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	Queue queue = session.createQueue(QUEUE_NAME);

	MessageConsumer consumer = session.createConsumer(queue);
	Message m;

	// Retrieve the initial pre-upgrade messages
	for (int i = 1; i <= 5; i++)
	{
	m = consumer.receive(getReceiveTimeout());
	assertThat("Failed to receive an expected message", m, is(notNullValue()));
	assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(i)));
	assertThat("Message content was not as expected",
	((TextMessage) m).getText(),
	is(equalTo(STRING_1024_256)));
	}
	for (int i = 1; i <= 5; i++)
	{
	m = consumer.receive(getReceiveTimeout());
	assertThat("Failed to receive an expected message", m, is(notNullValue()));
	assertThat("ID property did not match", m.getIntProperty("ID"), is(equalTo(i)));
	assertThat("Message content was not as expected", ((TextMessage) m).getText(), is((equalTo(STRING_1024))));
	}

	if (extraMessage)
	{
	//verify that the extra message is received
	m = consumer.receive(getReceiveTimeout());
	assertThat("Failed to receive an expected message", m, is(notNullValue()));
	assertThat("ID property did not match", m.getIntProperty(INDEX), is(equalTo(0)));
	}
	else
	{
	// Verify that no more messages are received
	m = consumer.receive(getReceiveTimeout());
	assertThat("No more messages should have been recieved", m, is(nullValue()));
	}

	consumer.close();
	session.close();
	}

	/**
	* Generates a string of a given length consisting of the sequence 0,1,2,..,9,0,1,2.
	*
	* @param length number of characters in the string
	* @return string sequence of the given length
	*/
	private static String generateString(int length)
	{
	char[] base_chars = new char[]{'0','1','2','3','4','5','6','7','8','9'};
	char[] chars = new char[length];
	for (int i = 0; i < (length); i++)
	{
	chars[i] = base_chars[i % 10];
	}
	return new String(chars);
	}

	}