QPID-5715: Move coalescing commiters into environment facades. Flush logs for not completed commit futures if commit task is stopped

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1601445 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 10 insertions, 38 deletions

diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
index fd196a28da..a2ef422046 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
@@ -472,8 +472,8 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
         assertTrue("CoalescingSync is not ON", virtualHost.isCoalescingSync());
 
         Map<String, Object> virtualHostAttributes = new HashMap<String,Object>();
-        virtualHostAttributes.put(BDBHAVirtualHost.LOCAL_TRANSACTION_SYNCRONIZATION_POLICY, "WRITE_NO_SYNC");
-        virtualHostAttributes.put(BDBHAVirtualHost.REMOTE_TRANSACTION_SYNCRONIZATION_POLICY, "SYNC");
+        virtualHostAttributes.put(BDBHAVirtualHost.LOCAL_TRANSACTION_SYNCHRONIZATION_POLICY, "WRITE_NO_SYNC");
+        virtualHostAttributes.put(BDBHAVirtualHost.REMOTE_TRANSACTION_SYNCHRONIZATION_POLICY, "SYNC");
         virtualHost.setAttributes(virtualHostAttributes);
 
         awaitForAttributeChange(virtualHost, BDBHAVirtualHostImpl.COALESCING_SYNC, false);
@@ -482,7 +482,7 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
         assertFalse("CoalescingSync is not OFF", virtualHost.isCoalescingSync());
         try
         {
-            virtualHost.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.LOCAL_TRANSACTION_SYNCRONIZATION_POLICY, "INVALID"));
+            virtualHost.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.LOCAL_TRANSACTION_SYNCHRONIZATION_POLICY, "INVALID"));
             fail("Invalid syncronization policy is set");
         }
         catch(IllegalArgumentException e)
@@ -492,7 +492,7 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
 
         try
         {
-            virtualHost.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.REMOTE_TRANSACTION_SYNCRONIZATION_POLICY, "INVALID"));
+            virtualHost.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.REMOTE_TRANSACTION_SYNCHRONIZATION_POLICY, "INVALID"));
             fail("Invalid syncronization policy is set");
         }
         catch(IllegalArgumentException e)
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
index b14332ecf6..67364ada35 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
@@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.apache.qpid.server.store.berkeleydb.Committer;
 import org.apache.qpid.server.store.berkeleydb.EnvironmentFacade;
 import org.apache.qpid.test.utils.QpidTestCase;
 import org.apache.qpid.test.utils.TestFileUtils;
@@ -169,29 +168,13 @@ public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
         ReplicatedEnvironmentFacade master = createMaster();
         assertEquals("Unexpected message store durability", TEST_DURABILITY, master.getMessageStoreTransactionDurability());
         assertEquals("Unexpected durability", TEST_DURABILITY, master.getDurability());
-        Committer committer = master.createCommitter(TEST_GROUP_NAME);
-        committer.start();
-
-        waitForCommitter(committer, true);
-
+        assertFalse("Coalescing syn before policy set to SYNC", master.isCoalescingSync());
+        master.setMessageStoreLocalTransactionSynchronizationPolicy(SyncPolicy.SYNC);
+        assertTrue("Coalescing syn after policy set to SYNC", master.isCoalescingSync());
         assertEquals("Unexpected message store durability after committer start", "NO_SYNC,NO_SYNC,SIMPLE_MAJORITY", master.getMessageStoreTransactionDurability().toString());
-
-        committer.stop();
-        waitForCommitter(committer, false);
-        assertEquals("Unexpected message store durability after committer stop", TEST_DURABILITY, master.getMessageStoreTransactionDurability());
-    }
-
-    public void testIsCoalescingSync() throws Exception
-    {
-        ReplicatedEnvironmentFacade master = createMaster();
-        assertEquals("Unexpected coalescing sync", false, master.isCoalescingSync());
-        Committer committer = master.createCommitter(TEST_GROUP_NAME);
-        committer.start();
-        waitForCommitter(committer, true);
-        assertEquals("Unexpected coalescing sync", true, master.isCoalescingSync());
-        committer.stop();
-        waitForCommitter(committer, false);
-        assertEquals("Unexpected coalescing sync", false, master.isCoalescingSync());
+        master.setMessageStoreLocalTransactionSynchronizationPolicy(SyncPolicy.WRITE_NO_SYNC);
+        assertEquals("Unexpected message store durability after committer stop", "WRITE_NO_SYNC,NO_SYNC,SIMPLE_MAJORITY", master.getMessageStoreTransactionDurability().toString());
+        assertFalse("Coalescing syn after policy set to WRITE_NO_SYNC", master.isCoalescingSync());
     }
 
     public void testGetNodeState() throws Exception
@@ -719,17 +702,6 @@ public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
         return addNode(TEST_NODE_NAME, TEST_NODE_HOST_PORT, TEST_DESIGNATED_PRIMARY, desiredState, stateChangeListener, replicationGroupListener);
     }
 
-    private void waitForCommitter(Committer committer, boolean expected) throws InterruptedException
-    {
-        int counter = 0;
-        while(committer.isStarted() != expected && counter < 100)
-        {
-            Thread.sleep(20);
-            counter++;
-        }
-        assertEquals("Committer is not in expected state", expected, committer.isStarted());
-    }
-
     private ReplicatedEnvironmentConfiguration createReplicatedEnvironmentConfiguration(String nodeName, String nodeHostPort, boolean designatedPrimary)
     {
         ReplicatedEnvironmentConfiguration node = mock(ReplicatedEnvironmentConfiguration.class);