Merge pull request #1604 from cherrylzhao/dev

for #1363 and #1238 Resolve performance test problem.
apache · Dec 12, 2018 · 64a39a0 · 64a39a0
2 parents 4a99bc4 + 95452a8
commit 64a39a0
Show file tree

Hide file tree

Showing 7 changed files with 95 additions and 11 deletions.
diff --git a/sharding-core/src/main/java/io/shardingsphere/core/rule/DataSourceParameter.java b/sharding-core/src/main/java/io/shardingsphere/core/rule/DataSourceParameter.java
@@ -42,7 +42,7 @@ public final class DataSourceParameter {
 
     private static final int DEFAULT_MAX_POOL_SIZE = 50;
 
-    private static final int DEFAULT_MIN_POOL_SIZE = 5;
+    private static final int DEFAULT_MIN_POOL_SIZE = 1;
 
     private ProxyPoolType proxyDatasourceType = ProxyPoolType.VENDOR;
 

diff --git a/.../src/main/java/io/shardingsphere/shardingjdbc/jdbc/adapter/AbstractConnectionAdapter.java b/.../src/main/java/io/shardingsphere/shardingjdbc/jdbc/adapter/AbstractConnectionAdapter.java
@@ -18,7 +18,7 @@
 package io.shardingsphere.shardingjdbc.jdbc.adapter;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.HashMultimap;
+import com.google.common.collect.LinkedHashMultimap;
 import com.google.common.collect.Multimap;
 import io.shardingsphere.core.constant.ConnectionMode;
 import io.shardingsphere.core.constant.transaction.TransactionOperationType;
@@ -28,10 +28,10 @@
 import io.shardingsphere.core.event.transaction.xa.XATransactionEvent;
 import io.shardingsphere.core.hint.HintManagerHolder;
 import io.shardingsphere.core.routing.router.masterslave.MasterVisitedManager;
+import io.shardingsphere.core.transaction.TransactionTypeHolder;
 import io.shardingsphere.shardingjdbc.jdbc.adapter.executor.ForceExecuteCallback;
 import io.shardingsphere.shardingjdbc.jdbc.adapter.executor.ForceExecuteTemplate;
 import io.shardingsphere.shardingjdbc.jdbc.unsupported.AbstractUnsupportedOperationConnection;
-import io.shardingsphere.core.transaction.TransactionTypeHolder;
 import io.shardingsphere.spi.root.RootInvokeHook;
 import io.shardingsphere.spi.root.SPIRootInvokeHook;
 import io.shardingsphere.spi.transaction.ShardingTransactionHandler;
@@ -60,7 +60,7 @@
 @Getter
 public abstract class AbstractConnectionAdapter extends AbstractUnsupportedOperationConnection {
 
-    private final Multimap<String, Connection> cachedConnections = HashMultimap.create();
+    private final Multimap<String, Connection> cachedConnections = LinkedHashMultimap.create();
 
     private boolean autoCommit = true;
 
@@ -186,10 +186,13 @@ public void execute(final Connection connection) throws SQLException {
                     connection.setAutoCommit(autoCommit);
                 }
             });
-        } else if (TransactionType.XA == transactionType) {
-            shardingTransactionHandler.doInTransaction(new XATransactionEvent(TransactionOperationType.BEGIN));
-        } else if (TransactionType.BASE == transactionType) {
-            shardingTransactionHandler.doInTransaction(new SagaTransactionEvent(TransactionOperationType.BEGIN, this));
+        }
+        if (!autoCommit) {
+            if (TransactionType.XA == transactionType) {
+                shardingTransactionHandler.doInTransaction(new XATransactionEvent(TransactionOperationType.BEGIN));
+            } else if (TransactionType.BASE == transactionType) {
+                shardingTransactionHandler.doInTransaction(new SagaTransactionEvent(TransactionOperationType.BEGIN, this));
+            }
         }
     }
 

diff --git a/...core/src/test/java/io/shardingsphere/shardingjdbc/jdbc/adapter/ConnectionAdapterTest.java b/...core/src/test/java/io/shardingsphere/shardingjdbc/jdbc/adapter/ConnectionAdapterTest.java
@@ -18,10 +18,15 @@
 package io.shardingsphere.shardingjdbc.jdbc.adapter;
 
 import com.google.common.collect.Multimap;
+import io.shardingsphere.core.constant.transaction.TransactionType;
+import io.shardingsphere.core.transaction.TransactionTypeHolder;
 import io.shardingsphere.shardingjdbc.common.base.AbstractShardingJDBCDatabaseAndTableTest;
 import io.shardingsphere.shardingjdbc.jdbc.core.connection.ShardingConnection;
+import io.shardingsphere.shardingjdbc.jdbc.core.datasource.FixedBaseShardingTransactionHandler;
+import io.shardingsphere.shardingjdbc.jdbc.core.datasource.FixedXAShardingTransactionHandler;
 import io.shardingsphere.shardingjdbc.jdbc.util.JDBCTestSQL;
 import lombok.SneakyThrows;
+import org.junit.After;
 import org.junit.Test;
 
 import java.lang.reflect.Field;
@@ -30,6 +35,7 @@
 import java.sql.SQLException;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -38,6 +44,13 @@ public final class ConnectionAdapterTest extends AbstractShardingJDBCDatabaseAnd
 
     private String sql = JDBCTestSQL.SELECT_GROUP_BY_USER_ID_SQL;
 
+    @After
+    public void tearDown() {
+        TransactionTypeHolder.clear();
+        FixedXAShardingTransactionHandler.getInvokes().clear();
+        FixedBaseShardingTransactionHandler.getInvokes().clear();
+    }
+
     @Test
     public void assertSetAutoCommit() throws SQLException {
         try (ShardingConnection actual = getShardingDataSource().getConnection()) {
@@ -57,6 +70,24 @@ private void assertAutoCommit(final ShardingConnection actual, final boolean aut
         }
     }
 
+    @Test
+    public void assertIgnoreAutoCommitForXA() throws SQLException {
+        TransactionTypeHolder.set(TransactionType.XA);
+        try (ShardingConnection actual = getShardingDataSource().getConnection()) {
+            actual.setAutoCommit(true);
+            assertNull(FixedXAShardingTransactionHandler.getInvokes().get("begin"));
+        }
+    }
+
+    @Test
+    public void assertIgnoreAutoCommitForBase() throws SQLException {
+        TransactionTypeHolder.set(TransactionType.BASE);
+        try (ShardingConnection actual = getShardingDataSource().getConnection()) {
+            actual.setAutoCommit(true);
+            assertNull(FixedBaseShardingTransactionHandler.getInvokes().get("begin"));
+        }
+    }
+
     @Test
     // TODO 缺少断言，做柔性事务时补充
     public void assertCommit() throws SQLException {
@@ -67,6 +98,15 @@ public void assertCommit() throws SQLException {
         }
     }
 
+    @Test
+    public void assertXACommit() throws SQLException {
+        TransactionTypeHolder.set(TransactionType.XA);
+        try (ShardingConnection actual = getShardingDataSource().getConnection()) {
+            actual.commit();
+            assertNotNull(FixedXAShardingTransactionHandler.getInvokes().get("commit"));
+        }
+    }
+
     @Test
     // TODO 缺少断言，做柔性事务时补充
     public void assertRollback() throws SQLException {
@@ -77,6 +117,15 @@ public void assertRollback() throws SQLException {
         }
     }
 
+    @Test
+    public void assertXARollback() throws SQLException {
+        TransactionTypeHolder.set(TransactionType.XA);
+        try (ShardingConnection actual = getShardingDataSource().getConnection()) {
+            actual.rollback();
+            assertNotNull(FixedXAShardingTransactionHandler.getInvokes().get("rollback"));
+        }
+    }
+
     @Test
     public void assertClose() throws SQLException {
         try (ShardingConnection actual = getShardingDataSource().getConnection()) {

diff --git a/.../java/io/shardingsphere/shardingproxy/backend/jdbc/connection/ConnectionStateHandler.java b/.../java/io/shardingsphere/shardingproxy/backend/jdbc/connection/ConnectionStateHandler.java
@@ -79,6 +79,9 @@ void doNotifyIfNecessary() {
         if (status.compareAndSet(ConnectionStatus.RUNNING, ConnectionStatus.RELEASE)) {
             resourceSynchronizer.doNotify();
         }
+        if (status.compareAndSet(ConnectionStatus.TERMINATED, ConnectionStatus.RELEASE)) {
+            resourceSynchronizer.doNotify();
+        }
     }
 
     /**
@@ -87,7 +90,7 @@ void doNotifyIfNecessary() {
      * @throws InterruptedException interrupted exception
      */
     public void waitUntilConnectionReleasedIfNecessary() throws InterruptedException {
-        if (ConnectionStatus.RUNNING == status.get()) {
+        if (ConnectionStatus.RUNNING == status.get() || ConnectionStatus.TERMINATED == status.get()) {
             while (!status.compareAndSet(ConnectionStatus.RELEASE, ConnectionStatus.RUNNING)) {
                 resourceSynchronizer.doAwait();
             }

diff --git a/...a/io/shardingsphere/shardingproxy/backend/jdbc/connection/ConnectionStateHandlerTest.java b/...a/io/shardingsphere/shardingproxy/backend/jdbc/connection/ConnectionStateHandlerTest.java
@@ -58,4 +58,33 @@ public void run() {
         notifyThread.join();
         assertTrue(flag.get());
     }
+
+    @Test
+    public void assertWaitUntilConnectionReleaseForTransaction() throws InterruptedException {
+        final AtomicBoolean flag = new AtomicBoolean(true);
+        Thread waitThread = new Thread(new Runnable() {
+            @Override
+            @SneakyThrows
+            public void run() {
+                connectionStateHandler.getAndSetStatus(ConnectionStatus.TERMINATED);
+                connectionStateHandler.waitUntilConnectionReleasedIfNecessary();
+                if (ConnectionStatus.RUNNING != connectionStateHandler.getStatus()) {
+                    flag.getAndSet(false);
+                }
+            }
+        });
+        Thread notifyThread = new Thread(new Runnable() {
+            @Override
+            @SneakyThrows
+            public void run() {
+                Thread.sleep(2000);
+                connectionStateHandler.doNotifyIfNecessary();
+            }
+        });
+        waitThread.start();
+        notifyThread.start();
+        waitThread.join();
+        notifyThread.join();
+        assertTrue(flag.get());
+    }
 }
diff --git a/sharding-transaction/sharding-transaction-xa/src/main/resources/jta.properties b/sharding-transaction/sharding-transaction-xa/src/main/resources/jta.properties
@@ -1,4 +1,4 @@
 com.atomikos.icatch.serial_jta_transactions = false
 com.atomikos.icatch.default_jta_timeout = 1000000
 com.atomikos.icatch.max_actives = 10000
-com.atomikos.icatch.enable_logging = false
+com.atomikos.icatch.enable_logging = true
diff --git a/...est/java/io/shardingsphere/transaction/xa/convert/swap/DataSourceSwapperRegistryTest.java b/...est/java/io/shardingsphere/transaction/xa/convert/swap/DataSourceSwapperRegistryTest.java
@@ -80,7 +80,7 @@ public void assertBuildParameterFromUnsupportedDataSource() {
         assertNull(actual.getUsername());
         assertNull(actual.getPassword());
         assertThat(actual.getMaximumPoolSize(), is(50));
-        assertThat(actual.getMinimumPoolSize(), is(5));
+        assertThat(actual.getMinimumPoolSize(), is(1));
         assertThat(actual.getConnectionTimeout(), is(30 * 1000L));
         assertThat(actual.getIdleTimeout(), is(60 * 1000L));
         assertThat(actual.getMaxLifetime(), is(0L));