Cluster: an open transaction was committed when a cluster node was stopped

(because disabling the cluster executes SET CLUSTER '', which committed the transaction).
Transaction are no longer committed when calling SET CLUSTER.
Cluster: non-admin users couldn't connect to the cluster and couldn't disable the cluster. Issue 201.

h2/src/main/org/h2/command/dml/Set.java

@@ -47,6 +47,7 @@ public class Set extends Prepared {
 
     public boolean isTransactional() {
         switch (type) {
+        case SetTypes.CLUSTER:
         case SetTypes.VARIABLE:
         case SetTypes.QUERY_TIMEOUT:
         case SetTypes.LOCK_TIMEOUT:
@@ -81,9 +82,17 @@ public class Set extends Prepared {
             addOrUpdateSetting(name, null, getIntValue());
             break;
         case SetTypes.CLUSTER: {
-            session.getUser().checkAdmin();
-            database.setCluster(StringUtils.quoteStringSQL(stringValue));
-            addOrUpdateSetting(name, StringUtils.quoteStringSQL(stringValue), 0);
+            String value = StringUtils.quoteStringSQL(stringValue);
+            if (!value.equals(database.getCluster()) && !value.equals("''")) {
+                // anybody can disable the cluster
+                // (if he can't access a cluster node)
+                session.getUser().checkAdmin();
+            }
+            database.setCluster(value);
+            // use the system session so that the current transaction
+            // (if any) is not committed
+            addOrUpdateSetting(database.getSystemSession(), name, value, 0);
+            database.getSystemSession().commit(true);
             break;
         }
         case SetTypes.COLLATION: {
@@ -352,6 +361,10 @@ public class Set extends Prepared {
     }
 
     private void addOrUpdateSetting(String name, String s, int v) {
+        addOrUpdateSetting(session, name, s, v);
+    }
+
+    private void addOrUpdateSetting(Session session, String name, String s, int v) {
         Database database = session.getDatabase();
         if (database.isReadOnly()) {
             return;

h2/src/test/org/h2/test/db/TestCluster.java

@@ -62,6 +62,8 @@ public class TestCluster extends TestBase {
         stat = conn.createStatement();
         stat.execute("create table test(id int primary key, name varchar) as " +
                 "select x, 'Data' || x from system_range(0, " + (len - 1) + ")");
+        stat.execute("create user test password 'test'");
+        stat.execute("grant all on test to test");
 
         // start the second server
         Server n2 = org.h2.tools.Server.createTcpServer("-tcpPort", "" + port2 , "-baseDir", getBaseDir() + "/node2").start();
@@ -83,9 +85,18 @@ public class TestCluster extends TestBase {
         Connection connApp = DriverManager.getConnection(urlCluster + ";AUTO_RECONNECT=TRUE", user, password);
         check(connApp, len, "'" + serverList + "'");
 
+        // delete the rows, but don't commit
+        connApp.setAutoCommit(false);
+        connApp.createStatement().execute("delete from test");
+
         // stop server 2, and test if only one server is available
         n2.stop();
+
+        // rollback the transaction
+        connApp.createStatement().executeQuery("select count(*) from test");
+        connApp.rollback();
         check(connApp, len, "''");
+        connApp.setAutoCommit(true);
 
         // re-create the cluster
         n2 = org.h2.tools.Server.createTcpServer("-tcpPort", "" + port2, "-baseDir", getBaseDir() + "/node2").start();
@@ -95,6 +106,11 @@ public class TestCluster extends TestBase {
         // test the cluster connection
         check(connApp, len, "'" + serverList + "'");
 
+        // test a non-admin user
+        String user2 = "test", password2 = getPassword("test");
+        connApp = DriverManager.getConnection(urlCluster, user2, password2);
+        check(connApp, len, "'" + serverList + "'");
+
         n1.stop();
         n2.stop();
         deleteFiles();