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 @@
    * 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 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 @@
       throw new RuntimeException(e);
     }
     this.maxRetriesPerExecute = getNumStatementRetries(info);
+    this.transparentReconnectEnabled = config().transparentReconnectionEnabled();
   }
 
   /** Computes the number of retries
@@ -792,7 +795,8 @@
         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 @@
   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 @@
   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 @@
         .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 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 @@
   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 @@
     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 @@
     }
   }
 
+  @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);