This is an automated email from the ASF dual-hosted git repository.
elserj pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/calcite-avatica.git
The following commit(s) were added to refs/heads/master by this push:
new c014f71 CALCITE-5009 Transparent JDBC connection re-creation may lead
to data loss
c014f71 is described below
commit c014f71438762d6b69a4dbefdd2b7f27b4af63fa
Author: Istvan Toth <[email protected]>
AuthorDate: Tue Mar 29 12:43:34 2022 +0200
CALCITE-5009 Transparent JDBC connection re-creation may lead to data loss
CALCITE-903 has introduced a transaprent reconnection feature, which will
open
a new server-side connection in case it is expired from the server side
connection
cache.
While this is convinient for most read-only analytical workload, this
can cause a number a problems, including data loss for transactional
connections.
This patch disables the transparent reconnect feature by default, and adds
the
transparent_reconnection property, which re-enables it when set to true.
---
.../apache/calcite/avatica/AvaticaConnection.java | 6 +++-
.../calcite/avatica/BuiltInConnectionProperty.java | 4 ++-
.../apache/calcite/avatica/ConnectionConfig.java | 2 ++
.../calcite/avatica/ConnectionConfigImpl.java | 5 +++
.../avatica/remote/ConnectionPropertiesTest.java | 41 +++++++++++++++++++++-
5 files changed, 55 insertions(+), 3 deletions(-)
diff --git
a/core/src/main/java/org/apache/calcite/avatica/AvaticaConnection.java
b/core/src/main/java/org/apache/calcite/avatica/AvaticaConnection.java
index 6a5830b..f740d85 100644
--- a/core/src/main/java/org/apache/calcite/avatica/AvaticaConnection.java
+++ b/core/src/main/java/org/apache/calcite/avatica/AvaticaConnection.java
@@ -66,6 +66,7 @@ public abstract class AvaticaConnection implements Connection
{
* the number of rows modified. */
public static final String ROWCOUNT_COLUMN_NAME = "ROWCOUNT";
+ //TODO shouldn't we move this to BuiltInConnectionProperty ?
public static final String NUM_EXECUTE_RETRIES_KEY =
"avatica.statement.retries";
public static final String NUM_EXECUTE_RETRIES_DEFAULT = "5";
@@ -96,6 +97,7 @@ public abstract class AvaticaConnection implements Connection
{
public final Map<Integer, AvaticaStatement> statementMap = new
ConcurrentHashMap<>();
final Map<Integer, AtomicBoolean> flagMap = new ConcurrentHashMap<>();
protected final long maxRetriesPerExecute;
+ protected final boolean transparentReconnectEnabled;
/**
* Creates an AvaticaConnection.
@@ -127,6 +129,7 @@ public abstract class AvaticaConnection implements
Connection {
throw new RuntimeException(e);
}
this.maxRetriesPerExecute = getNumStatementRetries(info);
+ this.transparentReconnectEnabled =
config().transparentReconnectionEnabled();
}
/** Computes the number of retries
@@ -792,7 +795,8 @@ public abstract class AvaticaConnection implements
Connection {
return callable.call();
} catch (AvaticaClientRuntimeException e) {
lastException = e;
- if (ErrorResponse.MISSING_CONNECTION_ERROR_CODE == e.getErrorCode()) {
+ if (ErrorResponse.MISSING_CONNECTION_ERROR_CODE == e.getErrorCode()
+ && transparentReconnectEnabled) {
this.openConnection();
continue;
}
diff --git
a/core/src/main/java/org/apache/calcite/avatica/BuiltInConnectionProperty.java
b/core/src/main/java/org/apache/calcite/avatica/BuiltInConnectionProperty.java
index 16e1061..0313e4c 100644
---
a/core/src/main/java/org/apache/calcite/avatica/BuiltInConnectionProperty.java
+++
b/core/src/main/java/org/apache/calcite/avatica/BuiltInConnectionProperty.java
@@ -86,7 +86,9 @@ public enum BuiltInConnectionProperty implements
ConnectionProperty {
KEY_PASSWORD("key_password", Type.STRING, "", false),
HOSTNAME_VERIFICATION("hostname_verification", Type.ENUM,
HostnameVerification.STRICT,
- HostnameVerification.class, false);
+ HostnameVerification.class, false),
+
+ TRANSPARENT_RECONNECTION("transparent_reconnection", Type.BOOLEAN,
Boolean.FALSE, false);
private final String camelName;
private final Type type;
diff --git
a/core/src/main/java/org/apache/calcite/avatica/ConnectionConfig.java
b/core/src/main/java/org/apache/calcite/avatica/ConnectionConfig.java
index bbbfa87..30d0c57 100644
--- a/core/src/main/java/org/apache/calcite/avatica/ConnectionConfig.java
+++ b/core/src/main/java/org/apache/calcite/avatica/ConnectionConfig.java
@@ -62,6 +62,8 @@ public interface ConnectionConfig {
String keyPassword();
/** @see BuiltInConnectionProperty#HOSTNAME_VERIFICATION */
HostnameVerification hostnameVerification();
+ /** @see BuiltInConnectionProperty#TRANSPARENT_RECONNECTION */
+ boolean transparentReconnectionEnabled();
}
// End ConnectionConfig.java
diff --git
a/core/src/main/java/org/apache/calcite/avatica/ConnectionConfigImpl.java
b/core/src/main/java/org/apache/calcite/avatica/ConnectionConfigImpl.java
index 36cdf61..7a2cbe4 100644
--- a/core/src/main/java/org/apache/calcite/avatica/ConnectionConfigImpl.java
+++ b/core/src/main/java/org/apache/calcite/avatica/ConnectionConfigImpl.java
@@ -128,6 +128,11 @@ public class ConnectionConfigImpl implements
ConnectionConfig {
.getEnum(HostnameVerification.class);
}
+ @Override public boolean transparentReconnectionEnabled() {
+ return BuiltInConnectionProperty.TRANSPARENT_RECONNECTION.wrap(properties)
+ .getBoolean();
+ }
+
/** Converts a {@link Properties} object containing (name, value)
* pairs into a map whose keys are
* {@link org.apache.calcite.avatica.InternalProperty} objects.
diff --git
a/server/src/test/java/org/apache/calcite/avatica/remote/ConnectionPropertiesTest.java
b/server/src/test/java/org/apache/calcite/avatica/remote/ConnectionPropertiesTest.java
index 9527dc5..515c8b9 100644
---
a/server/src/test/java/org/apache/calcite/avatica/remote/ConnectionPropertiesTest.java
+++
b/server/src/test/java/org/apache/calcite/avatica/remote/ConnectionPropertiesTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.calcite.avatica.remote;
+import org.apache.calcite.avatica.AvaticaClientRuntimeException;
import org.apache.calcite.avatica.AvaticaConnection;
+import org.apache.calcite.avatica.BuiltInConnectionProperty;
import org.apache.calcite.avatica.ConnectionSpec;
import org.apache.calcite.avatica.jdbc.JdbcMeta;
import org.apache.calcite.avatica.server.HttpServer;
@@ -36,6 +38,8 @@ import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/** Tests covering {@link ConnectionPropertiesTest}. */
@RunWith(Parameterized.class)
@@ -48,6 +52,7 @@ public class ConnectionPropertiesTest {
private final String url;
private final int port;
private final Driver.Serialization serialization;
+ private final Properties clientReconnectProps = new Properties();
@Parameterized.Parameters(name = "{0}")
public static List<Object[]> parameters() throws Exception {
@@ -63,13 +68,16 @@ public class ConnectionPropertiesTest {
this.port = this.server.getPort();
this.serialization = serialization;
this.url = SERVERS.getJdbcUrl(port, serialization);
+ this.clientReconnectProps.setProperty(
+ BuiltInConnectionProperty.TRANSPARENT_RECONNECTION.camelName(),
"true");
}
@Test
public void testConnectionPropertiesSync() throws Exception {
ConnectionSpec.getDatabaseLock().lock();
try {
- AvaticaConnection conn = (AvaticaConnection)
DriverManager.getConnection(url);
+ AvaticaConnection conn =
+ (AvaticaConnection) DriverManager.getConnection(url,
clientReconnectProps);
conn.setAutoCommit(false);
conn.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
@@ -97,6 +105,37 @@ public class ConnectionPropertiesTest {
}
}
+ @Test
+ public void testConnectionPropertiesSyncNoReconnect() throws Exception {
+ ConnectionSpec.getDatabaseLock().lock();
+ try {
+ AvaticaConnection conn = (AvaticaConnection)
DriverManager.getConnection(url);
+ conn.setAutoCommit(false);
+ conn.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
+
+ // sync connection properties
+ conn.createStatement();
+ Connection remoteConn = getConnection(
+
AvaticaServersForTest.PropertyRemoteJdbcMetaFactory.getInstance(PROPERTIES),
conn.id);
+
+ assertFalse(remoteConn.getAutoCommit());
+ assertEquals(remoteConn.getTransactionIsolation(),
+ Connection.TRANSACTION_REPEATABLE_READ);
+
+ // after 1s, remote connection expired and reopen
+ Thread.sleep(1000);
+
+ try {
+ conn.createStatement();
+ fail("Should have thrown AvaticaClientRuntimeException");
+ } catch (Exception e) {
+ assertTrue(e instanceof AvaticaClientRuntimeException);
+ }
+ } finally {
+ ConnectionSpec.getDatabaseLock().unlock();
+ }
+ }
+
private static Connection getConnection(JdbcMeta m, String id) throws
Exception {
Field f = JdbcMeta.class.getDeclaredField("connectionCache");
f.setAccessible(true);
