content: /*
   * Copyright (c) 2015, 2021, Oracle and/or its affiliates.
   *
   * This program is free software; you can redistribute it and/or modify it under
   * the terms of the GNU General Public License, version 2.0, as published by the
   * Free Software Foundation.
   *
   * This program is also distributed with certain software (including but not
   * limited to OpenSSL) that is licensed under separate terms, as designated in a
   * particular file or component or in included license documentation. The
   * authors of MySQL hereby grant you an additional permission to link the
   * program and your derivative works with the separately licensed software that
   * they have included with MySQL.
   *
   * Without limiting anything contained in the foregoing, this file, which is
   * part of MySQL Connector/J, is also subject to the Universal FOSS Exception,
   * version 1.0, a copy of which can be found at
   * http://oss.oracle.com/licenses/universal-foss-exception.
   *
   * This program is distributed in the hope that it will be useful, but WITHOUT
   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
   * FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
   * for more details.
   *
   * You should have received a copy of the GNU General Public License along with
   * this program; if not, write to the Free Software Foundation, Inc.,
   * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
   */
  
  package testsuite.x.devapi;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  import static org.junit.jupiter.api.Assumptions.assumeTrue;
  
  import java.io.FileNotFoundException;
  import java.lang.ref.WeakReference;
  import java.lang.reflect.Field;
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.Random;
  import java.util.Set;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  import com.google.protobuf.MessageLite;
  import com.mysql.cj.Constants;
  import com.mysql.cj.CoreSession;
  import com.mysql.cj.ServerVersion;
  import com.mysql.cj.conf.PropertyDefinitions;
  import com.mysql.cj.conf.PropertyDefinitions.Compression;
  import com.mysql.cj.conf.PropertyDefinitions.XdevapiSslMode;
  import com.mysql.cj.conf.PropertyKey;
  import com.mysql.cj.exceptions.CJCommunicationsException;
  import com.mysql.cj.exceptions.CJPacketTooBigException;
  import com.mysql.cj.exceptions.FeatureNotAvailableException;
  import com.mysql.cj.exceptions.MysqlErrorNumbers;
  import com.mysql.cj.exceptions.WrongArgumentException;
  import com.mysql.cj.protocol.SocketFactory;
  import com.mysql.cj.protocol.x.XProtocolError;
  import com.mysql.cj.x.protobuf.Mysqlx.ServerMessages;
  import com.mysql.cj.x.protobuf.MysqlxNotice.Frame;
  import com.mysql.cj.xdevapi.Client;
  import com.mysql.cj.xdevapi.Client.ClientProperty;
  import com.mysql.cj.xdevapi.ClientFactory;
  import com.mysql.cj.xdevapi.ClientImpl;
  import com.mysql.cj.xdevapi.ClientImpl.PooledXProtocol;
  import com.mysql.cj.xdevapi.Collection;
  import com.mysql.cj.xdevapi.DbDoc;
  import com.mysql.cj.xdevapi.DocResult;
  import com.mysql.cj.xdevapi.FindStatement;
  import com.mysql.cj.xdevapi.JsonNumber;
  import com.mysql.cj.xdevapi.JsonString;
  import com.mysql.cj.xdevapi.Row;
  import com.mysql.cj.xdevapi.RowResult;
  import com.mysql.cj.xdevapi.Schema;
  import com.mysql.cj.xdevapi.SelectStatement;
  import com.mysql.cj.xdevapi.Session;
  import com.mysql.cj.xdevapi.SessionFactory;
  import com.mysql.cj.xdevapi.SessionImpl;
  import com.mysql.cj.xdevapi.SqlMultiResult;
  import com.mysql.cj.xdevapi.SqlResult;
  import com.mysql.cj.xdevapi.SqlStatement;
  import com.mysql.cj.xdevapi.Warning;
  import com.mysql.cj.xdevapi.XDevAPIError;
  
  import testsuite.InjectedSocketFactory;
  import testsuite.UnreliableSocketFactory;
  
  public class SessionTest extends DevApiBaseTestCase {
      @BeforeEach
      public void setupSessionTest() {
          assumeTrue(this.isSetForXTests, PropertyDefinitions.SYSP_testsuite_url_mysqlx + " must be set to run this test.");
          setupTestSession();
      }
  
      @AfterEach
      public void teardownSessionTest() {
          if (this.isSetForXTests) {
              this.createdTestSchemas.forEach(schemaName -> {
                  try {
                      this.session.dropSchema(schemaName);
                  } catch (XProtocolError x) {
                      // ignored
                  }
              });
              destroyTestSession();
          }
      }
  
      private List<String> createdTestSchemas = new ArrayList<>();
  
      /**
       * Create a random schema name. The schema will be dropped upon test cleanup.
       * 
       * @return a string
       */
      private String getRandomTestSchemaName() {
          String n = "cj_test_schema_no_" + new Random().nextInt(1000);
          this.createdTestSchemas.add(n);
          return n;
      }
  
      @Test
      public void urlWithDefaultSchema() {
          try {
              // Create user with mysql_native_password authentication plugin as it can be used with any of the authentication mechanisms.
              this.session.sql("CREATE USER IF NOT EXISTS 'testUserN'@'%' IDENTIFIED WITH mysql_native_password BY 'testUserN'").execute();
              this.session.sql("GRANT SELECT ON *.* TO 'testUserN'@'%'").execute();
  
              final String testSchemaName = getRandomTestSchemaName();
              this.session.createSchema(testSchemaName);
  
              final SessionFactory testSessionFactory = new SessionFactory();
              final String testUriPattern = "mysqlx://testUserN:testUserN@%s:%s/%s?xdevapi.auth=%s";
  
              // Check if the default schema is correctly sent when using different authentication mechanisms.
              String[] authMechs = mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.4")) ? new String[] { "PLAIN", "MYSQL41", "SHA256_MEMORY" }
                      : new String[] { "PLAIN", "MYSQL41" };
              for (String authMech : authMechs) {
                  final String testCase = "Testing default schema provided in authentication mecanism '" + authMech + "'.";
  
                  // Test using a connection String.
                  final String testUri = String.format(testUriPattern, getTestHost(), getTestPort(), testSchemaName, authMech);
  
                  Session testSession = testSessionFactory.getSession(testUri);
                  assertTrue(testSession.getUri().contains("/" + testSchemaName + "?"), testCase);
                  assertEquals(testSchemaName, testSession.getDefaultSchemaName(), testCase);
                  assertNotNull(testSession.getDefaultSchema(), testCase);
                  assertEquals(testSchemaName, testSession.getDefaultSchema().getName(), testCase);
                  assertEquals(testSchemaName, testSession.sql("SELECT database()").execute().fetchOne().getString(0), testCase);
                  testSession.close();
  
                  // Test using a properties map.
                  final Properties testProps = new Properties();
                  testProps.setProperty(PropertyKey.USER.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.PASSWORD.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.HOST.getKeyName(), getTestHost());
                  testProps.setProperty(PropertyKey.PORT.getKeyName(), String.valueOf(getTestPort()));
                  testProps.setProperty(PropertyKey.DBNAME.getKeyName(), testSchemaName);
                  testProps.setProperty(PropertyKey.xdevapiAuth.getKeyName(), authMech);
  
                  testSession = testSessionFactory.getSession(testProps);
                  assertTrue(testSession.getUri().contains("/" + testSchemaName + "?"), testCase);
                  assertEquals(testSchemaName, testSession.getDefaultSchemaName(), testCase);
                  assertNotNull(testSession.getDefaultSchema(), testCase);
                  assertEquals(testSchemaName, testSession.getDefaultSchema().getName(), testCase);
                  assertEquals(testSchemaName, testSession.sql("SELECT database()").execute().fetchOne().getString(0), testCase);
                  testSession.close();
              }
          } finally {
              this.session.sql("DROP USER IF EXISTS testUserN").execute();
          }
      }
  
      @Test
      public void urlWithoutDefaultSchema() {
          try {
              // Create user with mysql_native_password authentication plugin as it can be used with any of the authentication mechanisms.
              this.session.sql("CREATE USER IF NOT EXISTS 'testUserN'@'%' IDENTIFIED WITH mysql_native_password BY 'testUserN'").execute();
              this.session.sql("GRANT SELECT ON *.* TO 'testUserN'@'%'").execute();
  
              final SessionFactory testSessionFactory = new SessionFactory();
              final String testUriPattern1 = "mysqlx://testUserN:testUserN@%s:%s/?xdevapi.auth=%s";
              final String testUriPattern2 = "mysqlx://testUserN:testUserN@%s:%s?xdevapi.auth=%s";
              final String testUriPattern3 = "mysqlx://testUserN:testUserN@address=(host=%s)(port=%s)(xdevapi.auth=%s)";
  
              // Check if not setting a default schema works correctly when using different authentication mechanisms.
              String[] authMechs = mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.4")) ? new String[] { "PLAIN", "MYSQL41", "SHA256_MEMORY" }
                      : new String[] { "PLAIN", "MYSQL41" };
              for (String authMech : authMechs) {
                  for (String testUriPattern : new String[] { testUriPattern1, testUriPattern2, testUriPattern3 }) {
                      // Test using a connection String.
                      final String testUri = String.format(testUriPattern, getTestHost(), getTestPort(), authMech);
                      final String testCase = "Testing no default schema with authentication mecanism '" + authMech + "' and URI '" + testUri + "'.";
  
                      Session testSession = testSessionFactory.getSession(testUri);
                      assertTrue(testSession.getUri().contains("/?"), testCase);
                      assertEquals("", testSession.getDefaultSchemaName(), testCase);
                      assertNull(testSession.getDefaultSchema(), testCase);
                      assertNull(testSession.sql("SELECT database()").execute().fetchOne().getString(0), testCase);
                      testSession.close();
                  }
  
                  // Test using a properties map.
                  final String testCase = "Testing no default schema with authentication mecanism '" + authMech + "'.";
                  final Properties testProps = new Properties();
                  testProps.setProperty(PropertyKey.USER.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.PASSWORD.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.HOST.getKeyName(), getTestHost());
                  testProps.setProperty(PropertyKey.PORT.getKeyName(), String.valueOf(getTestPort()));
                  testProps.setProperty(PropertyKey.xdevapiAuth.getKeyName(), authMech);
  
                  Session testSession = testSessionFactory.getSession(testProps);
                  assertTrue(testSession.getUri().contains("/?"), testCase);
                  assertEquals("", testSession.getDefaultSchemaName(), testCase);
                  assertNull(testSession.getDefaultSchema(), testCase);
                  assertNull(testSession.sql("SELECT database()").execute().fetchOne().getString(0), testCase);
                  testSession.close();
              }
          } finally {
              this.session.sql("DROP USER IF EXISTS testUserN").execute();
          }
      }
  
      @Test
      public void invalidDefaultSchema() {
          try {
              // Create user with mysql_native_password authentication plugin as it can be used with any of the authentication mechanisms.
              this.session.sql("CREATE USER IF NOT EXISTS 'testUserN'@'%' IDENTIFIED WITH mysql_native_password BY 'testUserN'").execute();
              this.session.sql("GRANT SELECT ON *.* TO 'testUserN'@'%'").execute();
  
              final String testSchemaName = getRandomTestSchemaName();
  
              final SessionFactory testSessionFactory = new SessionFactory();
              final String testUriPattern = "mysqlx://testUserN:testUserN@%s:%s/%s?xdevapi.auth=%s";
  
              // Check if the default schema is correctly sent when using different authentication mechanisms.
              String[] authMechs = mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.4")) ? new String[] { "PLAIN", "MYSQL41", "SHA256_MEMORY" }
                      : new String[] { "PLAIN", "MYSQL41" };
              for (String authMech : authMechs) {
                  final String testCase = "Testing missing default schema provided in authentication mecanism '" + authMech + "'.";
  
                  // Test using a connection String.
                  final String testUri = String.format(testUriPattern, getTestHost(), getTestPort(), testSchemaName, authMech);
  
                  assertThrows(testCase, XProtocolError.class, "ERROR \\d{4} \\(HY000\\) Unknown database '" + testSchemaName + "'", () -> {
                      testSessionFactory.getSession(testUri);
                      return null;
                  });
  
                  // Test using a properties map.
                  final Properties testProps = new Properties();
                  testProps.setProperty(PropertyKey.USER.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.PASSWORD.getKeyName(), "testUserN");
                  testProps.setProperty(PropertyKey.HOST.getKeyName(), getTestHost());
                  testProps.setProperty(PropertyKey.PORT.getKeyName(), String.valueOf(getTestPort()));
                  testProps.setProperty(PropertyKey.DBNAME.getKeyName(), testSchemaName);
                  testProps.setProperty(PropertyKey.xdevapiAuth.getKeyName(), authMech);
  
                  assertThrows(testCase, XProtocolError.class, "ERROR \\d{4} \\(HY000\\) Unknown database '" + testSchemaName + "'", () -> {
                      testSessionFactory.getSession(testUri);
                      return null;
                  });
              }
          } finally {
              this.session.sql("DROP USER IF EXISTS testUserN").execute();
          }
      }
  
      @Test
      public void createDropSchema() {
          String testSchemaName = getRandomTestSchemaName();
          Schema newSchema = this.session.createSchema(testSchemaName);
          assertTrue(this.session.getSchemas().contains(newSchema));
          this.session.dropSchema(testSchemaName);
          assertFalse(this.session.getSchemas().contains(newSchema));
      }
  
      @Test
      public void createAndReuseExistingSchema() {
          String testSchemaName = getRandomTestSchemaName();
          Schema newSchema = this.session.createSchema(testSchemaName);
          assertTrue(this.session.getSchemas().contains(newSchema));
          Schema reusedSchema = this.session.createSchema(testSchemaName, true);
          assertTrue(this.session.getSchemas().contains(reusedSchema));
      }
  
      @Test
      public void listSchemas() {
          List<Schema> schemas = this.session.getSchemas();
          // we should have visibility of at least these two
          Schema infoSchema = this.session.getSchema("information_schema");
          assertTrue(schemas.contains(infoSchema));
          Schema testSchema = this.session.getSchema(getTestDatabase());
          assertTrue(schemas.contains(testSchema));
      }
  
      @Test
      public void createExistingSchemaError() {
          String testSchemaName = getRandomTestSchemaName();
          Schema newSchema = this.session.createSchema(testSchemaName);
          assertTrue(this.session.getSchemas().contains(newSchema));
          try {
              this.session.createSchema(testSchemaName);
              fail("Attempt to create a schema with the name of an existing schema should fail");
          } catch (XProtocolError err) {
              assertEquals(MysqlErrorNumbers.ER_DB_CREATE_EXISTS, err.getErrorCode());
          }
      }
  
      /**
       * Test the client-side enforcing of server `mysqlx_max_allowed_packet'.
       */
      @Test
      public void errorOnPacketTooBig() {
          System.gc(); // to free the memory from previous tests artifacts
          SqlStatement stmt = this.session.sql("select @@mysqlx_max_allowed_packet");
          SqlResult res = stmt.execute();
          Row r = res.next();
          long mysqlxMaxAllowedPacket = r.getLong(0);
  
          long size = 100 + mysqlxMaxAllowedPacket;
          StringBuilder b = new StringBuilder();
          for (int i = 0; i < size; ++i) {
              b.append('a');
          }
          String s = b.append("\"}").toString();
          assertThrows("Large packet should cause an exception", CJPacketTooBigException.class, () -> {
              this.session.dropSchema(s);
              return null;
          });
      }
  
      /**
       * Tests fix for Bug#21690043, CONNECT FAILS WHEN PASSWORD IS BLANK.
       */
      @Test
      public void testBug21690043() {
          try {
              this.session.sql("CREATE USER 'bug21690043user1'@'%' IDENTIFIED WITH mysql_native_password").execute();
              this.session.sql("GRANT SELECT ON *.* TO 'bug21690043user1'@'%'").execute();
  
              Properties props = new Properties();
              props.putAll(this.testProperties);
              props.setProperty("user", "bug21690043user1");
              props.setProperty("password", "");
              new SessionFactory().getSession(props);
          } catch (Throwable t) {
              throw t;
          } finally {
              this.session.sql("DROP USER 'bug21690043user1'@'%'").execute();
          }
      }
  
      @Test
      public void basicSql() {
          SqlStatement stmt = this.session.sql("select 1,2,3 from dual");
          SqlResult res = stmt.execute();
          assertTrue(res.hasData());
          Row r = res.next();
          assertEquals("1", r.getString(0));
          assertEquals("2", r.getString(1));
          assertEquals("3", r.getString(2));
          assertEquals("1", r.getString("1"));
          assertEquals("2", r.getString("2"));
          assertEquals("3", r.getString("3"));
          assertFalse(res.hasNext());
  
          assertThrows(XDevAPIError.class, "Method getAutoIncrementValue\\(\\) is allowed only for insert statements.", new Callable<Void>() {
              public Void call() throws Exception {
                  assertEquals(null, res.getAutoIncrementValue());
                  return null;
              }
          });
      }
  
      @Test
      public void sqlUpdate() {
          SqlStatement stmt = this.session.sql("set @cjTestVar = 1");
          SqlResult res = stmt.execute();
          assertFalse(res.hasData());
          assertEquals(0, res.getAffectedItemsCount());
          assertEquals(null, res.getAutoIncrementValue());
          assertEquals(0, res.getWarningsCount());
          assertFalse(res.getWarnings().hasNext());
  
          // TODO SqlUpdateResult throws FeatureNotAvailableException("Not a multi-result");
          //res.nextResult();
  
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.fetchAll();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.next();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.hasNext();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.getColumnCount();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.getColumns();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.getColumnNames();
                  return null;
              }
          });
          assertThrows(FeatureNotAvailableException.class, "No data", new Callable<Void>() {
              public Void call() throws Exception {
                  res.count();
                  return null;
              }
          });
      }
  
      @Test
      public void sqlArguments() {
          SqlStatement stmt = this.session.sql("select ? as a, 40 + ? as b, ? as c");
          SqlResult res = stmt.bind(1).bind(2).bind(3).execute();
          Row r = res.next();
          assertEquals("1", r.getString("a"));
          assertEquals("42", r.getString("b"));
          assertEquals("3", r.getString("c"));
      }
  
      @Test
      public void basicMultipleResults() {
          try {
              sqlUpdate("drop procedure if exists basicMultipleResults");
              sqlUpdate("create procedure basicMultipleResults() begin explain select 1; explain select 2; end");
              SqlStatement stmt = this.session.sql("call basicMultipleResults()");
              SqlResult res = stmt.execute();
              assertTrue(res.hasData());
              /* Row r = */ res.next();
              assertFalse(res.hasNext());
              assertTrue(res.nextResult());
              assertTrue(res.hasData());
              assertFalse(res.nextResult());
              assertFalse(res.nextResult());
              assertFalse(res.nextResult());
          } finally {
              sqlUpdate("drop procedure if exists basicMultipleResults");
          }
      }
  
      @Test
      public void smartBufferMultipleResults() {
          try {
              sqlUpdate("drop procedure if exists basicMultipleResults");
              sqlUpdate("create procedure basicMultipleResults() begin explain select 1; explain select 2; end");
              SqlStatement stmt = this.session.sql("call basicMultipleResults()");
              /* SqlResult res = */ stmt.execute();
              // execute another statement, should work fine
              this.session.sql("call basicMultipleResults()");
              this.session.sql("call basicMultipleResults()");
              this.session.sql("call basicMultipleResults()");
          } finally {
              sqlUpdate("drop procedure if exists basicMultipleResults");
          }
      }
  
      @Test
      public void sqlInsertAutoIncrementValue() {
          try {
              sqlUpdate("drop table if exists lastInsertId");
              sqlUpdate("create table lastInsertId (id int not null primary key auto_increment, name varchar(20) not null)");
  
              SqlStatement stmt = this.session.sql("insert into lastInsertId values (null, 'a')");
              SqlResult res = stmt.execute();
  
              assertFalse(res.hasData());
              assertEquals(1, res.getAffectedItemsCount());
              assertEquals(0, res.getWarningsCount());
              assertFalse(res.getWarnings().hasNext());
              assertEquals(new Long(1), res.getAutoIncrementValue());
          } finally {
              sqlUpdate("drop table if exists lastInsertId");
          }
      }
  
      /**
       * Tests fix for Bug #27652379, NPE FROM GETSESSION(PROPERTIES) WHEN HOST PARAMETER IS GIVEN IN SMALL LETTER.
       * 
       * @throws Exception
       */
      @Test
      public void testBug27652379() throws Exception {
          Properties props = new Properties();
  
          // Upper case keys.
          props.clear();
          props.setProperty("HOST", getTestHost());
          props.setProperty("PORT", String.valueOf(getTestPort()));
          if (getTestUser() != null) {
              props.setProperty("USER", getTestUser());
          }
          if (getTestPassword() != null) {
              props.setProperty("PASSWORD", getTestPassword());
          }
          props.setProperty("DBNAME", getTestDatabase());
  
          Session testSession = this.fact.getSession(props);
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
  
          testSession = this.fact.getSession(new Properties(props));
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
  
          // Lower case keys.
          props.clear();
          props.setProperty("host", getTestHost());
          props.setProperty("port", String.valueOf(getTestPort()));
          if (getTestUser() != null) {
              props.setProperty("user", getTestUser());
          }
          if (getTestPassword() != null) {
              props.setProperty("password", getTestPassword());
          }
          props.setProperty("dbname", getTestDatabase());
  
          testSession = this.fact.getSession(props);
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
  
          testSession = this.fact.getSession(new Properties(props));
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
  
          // Random case keys.
          props.clear();
          props.setProperty("HOst", getTestHost());
          props.setProperty("poRT", String.valueOf(getTestPort()));
          if (getTestUser() != null) {
              props.setProperty("uSEr", getTestUser());
          }
          if (getTestPassword() != null) {
              props.setProperty("PassworD", getTestPassword());
          }
          props.setProperty("DbNaMe", getTestDatabase());
  
          testSession = this.fact.getSession(props);
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
  
          testSession = this.fact.getSession(new Properties(props));
          assertEquals(getTestDatabase(), testSession.getDefaultSchemaName());
          testSession.close();
      }
  
      /**
       * Tests fix for BUG#23045604 - XSESSION.GETURI() RETURNS NPE.
       */
      @Test
      public void testBug23045604() {
          String url = this.baseUrl;
          if (!url.contains("?")) {
              url += "?";
          }
          Session sess = this.fact
                  .getSession(url + makeParam(PropertyKey.connectionTimeZone, "Asia/Calcutta") + makeParam(PropertyKey.serverConfigCacheFactory, ""));
  
          String uri = sess.getUri();
          assertTrue(uri.contains("connectionTimeZone=Asia/Calcutta"));
          assertTrue(uri.contains("serverConfigCacheFactory="));
          assertFalse(uri.contains(","));
          sess.close();
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void testPooledSessions() throws Exception {
          final ClientFactory cf = new ClientFactory();
          final String url = this.baseUrl;
          final Properties props = new Properties();
  
          /*
           * UT8/1: Verify that setting an incorrect value in the client options defined in the Pooling options in the HLS throw an exception with the expected
           * message.
           */
          // pooling.enabled
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.enabled' does not support value 'sure'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty(ClientProperty.POOLING_ENABLED.getKeyName(), "sure");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.enabled' does not support value 'sure'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"enabled\": \"sure\"}}");
                  return null;
              }
          });
          // pooling.maxSize
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxSize' does not support value '0'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "0");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxSize' does not support value '0'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"maxSize\": 0}}");
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxSize' does not support value 'one'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"maxSize\": \"one\"}}");
                  return null;
              }
          });
          // pooling.maxIdleTime
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxIdleTime' does not support value '-1'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "-1");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxIdleTime' does not support value '-1'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"maxIdleTime\": -1}}");
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.maxIdleTime' does not support value 'one'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"maxIdleTime\": \"one\"}}");
                  return null;
              }
          });
          // pooling.queueTimeout
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.queueTimeout' does not support value '-1'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "-1");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.queueTimeout' does not support value '-1'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"queueTimeout\": -1}}");
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.queueTimeout' does not support value 'one'\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"queueTimeout\": \"one\"}}");
                  return null;
              }
          });
          // Unknown pooling option.
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.foo' is not recognized as valid\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty("pooling.foo", "bar");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'pooling\\.foo' is not recognized as valid\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"pooling\": {\"foo\": \"bar\"}}");
                  return null;
              }
          });
          // Unknown clientProps option.
          props.clear();
          assertThrows(XDevAPIError.class, "Client option 'foo' is not recognized as valid\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  props.setProperty("foo", "bar");
                  cf.getClient(url, props);
                  return null;
              }
          });
          assertThrows(XDevAPIError.class, "Client option 'foo' is not recognized as valid\\.", new Callable<Void>() {
              public Void call() throws Exception {
                  cf.getClient(url, "{\"foo\": {\"bar\": \"baz\"}}");
                  return null;
              }
          });
  
          /*
           * UT9/1: Verify that when no pooling properties passed their values set to defaults: pooling.maxSize=25, pooling.maxIdleTime=0, pooling.queueTimeout=0.
           */
          props.clear();
          testPooledSessions_checkClientProperties(cf.getClient(this.baseUrl, props), 25, 0, 0);
          testPooledSessions_checkClientProperties(cf.getClient(this.baseUrl, (Properties) null), 25, 0, 0);
          testPooledSessions_checkClientProperties(cf.getClient(this.baseUrl, "{}"), 25, 0, 0);
          testPooledSessions_checkClientProperties(cf.getClient(this.baseUrl, (String) null), 25, 0, 0);
  
          /*
           * UT9/2: Verify that when all pooling properties passed via Properties object the Client is configured according to their values.
           */
          props.setProperty(ClientProperty.POOLING_ENABLED.getKeyName(), "true");
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "5");
          props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "6");
          props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "7");
          testPooledSessions_checkClientProperties(cf.getClient(this.baseUrl, props), 5, 6, 7);
  
          /*
           * UT9/3: Verify that when all pooling properties passed via json string ({"pooling" : {"enabled" : true, "maxSize" : 8, "maxIdleTime" : 9,
           * "queueTimeout" : 10} }) the Client is configured according to their values.
           */
          testPooledSessions_checkClientProperties(
                  cf.getClient(this.baseUrl, "{\"pooling\" : {\"enabled\" : true, \"maxSize\" : 8, \"maxIdleTime\" : 9, \"queueTimeout\" : 10} }"), 8, 9, 10);
  
          /*
           * UT3/2: Start a client with pooling enabled, call Client.getSession() twice and verify that the objects returned are different and the internal
           * connection instances are different too.
           */
          Client cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true}}");
          Session s0 = cli0.getSession();
          Session s1 = cli0.getSession();
          assertNotEquals(s0, s1);
  
          Field fProtocol = CoreSession.class.getDeclaredField("protocol");
          fProtocol.setAccessible(true);
          assertNotEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT3/3: Start a client with pooling enabled, call Client.getSession() twice, closing the first session before getting the second one, and verify that
           * the objects returned are different and that the internal connection instances are the same.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true}}");
          s0 = cli0.getSession();
          s0.close();
          s1 = cli0.getSession();
          s1.sql("SELECT 1").execute();
          assertNotEquals(s0, s1);
          assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT3/4: Start a client with pooling disabled, call Client.getSession() twice and verify that the objects returned are different and the internal
           * connection instances are different too.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": false}}");
          s0 = cli0.getSession();
          s1 = cli0.getSession();
          assertNotEquals(s0, s1);
          assertNotEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT3/5: Start a client with pooling disabled, call Client.getSession() twice, closing the first session before getting the second one, and verify that
           * the objects returned are different and that the internal connection instances are different too.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": false}}");
          s0 = cli0.getSession();
          s0.close();
          s1 = cli0.getSession();
          assertNotEquals(s0, s1);
          assertNotEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT5/1: Having a full pool and queueTimeout = 0, verify that a new Client.getSession() waits until a Session is released.
           * UT6/1: Verify that that a client object can not open/create more sessions that the specified in the maxSize option.
           */
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "3");
          props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "1000");
          props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "1000");
          Client cli1 = cf.getClient(this.baseUrl, props);
          s1 = cli1.getSession();
          Session s2 = cli1.getSession();
          Session s3 = cli1.getSession();
          testPooledSessions_assertFailureTimeout(cli1, 1000, 2000, XDevAPIError.class, "Session can not be obtained within 1000 milliseconds.");
          cli1.close();
  
          /*
           * UT12/1: Having a pool with a single connection, close the Session, get another Session with Client.getSession(). Verify that the received Session
           * object is new and uses the same internal connection (MysqlxSession) instance.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 1}}");
          s0 = cli0.getSession();
          s0.close();
          s1 = cli0.getSession();
          s1.sql("SELECT 1").execute();
          assertNotEquals(s0, s1);
          assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT12/2: Having a pool with a number of connections greater than 1 and lower than maxPoolSize, close on Session, get another Session with
           * Client.getSession(). Verify that the received Session object is new and uses the same internal connection (MysqlxSession) instance.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 3}}");
          s0 = cli0.getSession();
          s0.close();
          s1 = cli0.getSession();
          s1.sql("SELECT 1").execute();
          assertNotEquals(s0, s1);
          assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
          cli0.close();
  
          /*
           * UT12/3: Having a full pool with all sessions in active state, close one Session, get another Session with Client.getSession(). Verify that the
           * received Session object is new and uses the same internal connection (MysqlxSession) instance.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 3}}");
          s0 = cli0.getSession();
          s1 = cli0.getSession();
          s2 = cli0.getSession();
          s0.close();
          s3 = cli0.getSession();
          s3.sql("SELECT 1").execute();
          assertNotEquals(s0, s3);
          assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s3).getSession()));
          cli0.close();
  
          /*
           * UT11/2: Having a pool with a single idle connection and maxIdleTime = n, verify that after n milliseconds a Client.getSession() call will remove all
           * expired sessions from pool and return a new Session object that uses a new MysqlxSession instance.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 3, \"maxIdleTime\" : 1000}}");
          s0 = cli0.getSession();
          s0.close();
          Thread.sleep(2000);
          s1 = cli0.getSession();
          assertNotEquals(s0, s1);
          assertNotEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
  
          Field fIdleSessions = ClientImpl.class.getDeclaredField("idleProtocols");
          fIdleSessions.setAccessible(true);
          assertTrue(((BlockingQueue<PooledXProtocol>) fIdleSessions.get(cli0)).isEmpty());
  
          /*
           * UT4/1: Verify that all idle and active sessions are closed after Client.close() call.
           */
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 3}}");
          s0 = cli0.getSession();
          s1 = cli0.getSession();
          s0.close();
  
          Field fActiveSessions = ClientImpl.class.getDeclaredField("activeProtocols");
          fActiveSessions.setAccessible(true);
          assertEquals(1, ((BlockingQueue<PooledXProtocol>) fIdleSessions.get(cli0)).size());
          assertEquals(1, ((Set<WeakReference<PooledXProtocol>>) fActiveSessions.get(cli0)).size());
  
          cli0.close();
          assertEquals(0, ((BlockingQueue<PooledXProtocol>) fIdleSessions.get(cli0)).size());
          assertEquals(0, ((Set<WeakReference<PooledXProtocol>>) fActiveSessions.get(cli0)).size());
  
          final Session ses = s1;
          assertThrows(CJCommunicationsException.class, new Callable<Void>() {
              public Void call() throws Exception {
                  ses.getSchemas();
                  return null;
              }
          });
  
          /*
           * UT4/2: Verify that after Client was closed the Client.getSession() throws an XDevAPIError with the message "Client is closed."
           */
          Client cli2 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true}}");
          cli2.close();
          assertThrows(XDevAPIError.class, "Client is closed.", new Callable<Void>() {
              public Void call() throws Exception {
                  cli2.getSession();
                  return null;
              }
          });
  
          /*
           * UT11/1: Having a pool with a single active and maxIdleTime = 0, verify that if closing the session then after any long inactivity time a new Session
           * object returned by Client.getSession() uses the same internal MysqlxSession object.
           */
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "2");
          props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "0");
          props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "20000");
          Client cli3 = cf.getClient(this.baseUrl, props);
          s0 = cli3.getSession();
          s1 = cli3.getSession();
          assertEquals(2, ((Set<WeakReference<PooledXProtocol>>) fActiveSessions.get(cli3)).size());
          s0.close();
          Thread.sleep(10000);
          s2 = cli3.getSession();
          s2.sql("SELECT 1").execute();
          assertNotEquals(s0, s2);
          assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s2).getSession()));
          cli3.close();
  
          /*
           * UT5/1: Having a full pool and queueTimeout = 0, verify that a new Client.getSession() waits until a Session is released.
           */
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "2");
          props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "1000");
          props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "0");
          cli3 = cf.getClient(this.baseUrl, props);
          s1 = cli3.getSession();
          Session s6 = cli3.getSession();
  
          Thread t = new Thread(new Runnable() {
              @Override
              public void run() {
                  try {
                      Thread.sleep(10000);
                      s6.close();
                  } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                  }
  
              }
          });
  
          long begin = System.currentTimeMillis();
          t.start();
          s1 = cli3.getSession();
          long end = System.currentTimeMillis() - begin;
          s1.sql("SELECT 1").execute();
          assertTrue(end >= 10000, "Expected wait time 10000 ms but was " + end);
          assertEquals(fProtocol.get(((SessionImpl) s1).getSession()), fProtocol.get(((SessionImpl) s6).getSession()));
  
          /*
           * TS10_2 Verify that RuntimProperty objects are reset to initial values when Session returned to pool.
           */
          assertEquals(2, ((Set<WeakReference<PooledXProtocol>>) fActiveSessions.get(cli3)).size());
          ((SessionImpl) s1).getSession().getPropertySet().getStringProperty(PropertyKey.connectionAttributes).setValue("orig:s1");
          s1.close();
          assertEquals(1, ((Set<WeakReference<PooledXProtocol>>) fActiveSessions.get(cli3)).size());
          s2 = cli3.getSession();
          s2.sql("SELECT 1").execute();
          assertEquals(fProtocol.get(((SessionImpl) s1).getSession()), fProtocol.get(((SessionImpl) s2).getSession()));
          assertNotEquals("orig:s1", ((SessionImpl) s1).getSession().getPropertySet().getStringProperty(PropertyKey.connectionAttributes).getValue());
          assertEquals(((SessionImpl) s1).getSession().getPropertySet().getStringProperty(PropertyKey.connectionAttributes).getInitialValue(),
                  ((SessionImpl) s1).getSession().getPropertySet().getStringProperty(PropertyKey.connectionAttributes).getValue());
  
          /*
           * UT10/1: Get the max sessions allowed in the pool, create the same variables and temp tables in all the sessions,
           * save ((SessionImpl) sN).getSession().getServerSession().getThreadId() values.
           * Then close all the sessions and get a new one: the variables and the temp tables must not exist,
           * ((SessionImpl) sN).getSession().getServerSession().getThreadId() must return values equal to saved ones.
           */
  
          cli0 = cf.getClient(this.baseUrl, "{\"pooling\": {\"enabled\": true, \"maxSize\" : 2}}");
          s0 = cli0.getSession();
          s1 = cli0.getSession();
          long id0 = ((SessionImpl) s0).getSession().getServerSession().getCapabilities().getThreadId();
          long id1 = ((SessionImpl) s1).getSession().getServerSession().getCapabilities().getThreadId();
  
          s0.sql("SET @a='s0'").execute();
          s0.sql("CREATE TEMPORARY TABLE testpooledsessionstmps0(x int)").execute();
  
          s1.sql("SET @a='s1'").execute();
          s1.sql("CREATE TEMPORARY TABLE testpooledsessionstmps1(x int)").execute();
  
          SqlResult res = s0.sql("SELECT @a as a").execute();
          assertTrue(res.hasNext());
          assertEquals("s0", res.next().getString(0));
          res = s0.sql("SHOW CREATE TABLE testpooledsessionstmps0").execute();
          assertTrue(res.hasNext());
          assertEquals("testpooledsessionstmps0", res.next().getString(0));
  
          res = s1.sql("SELECT @a as a").execute();
          assertTrue(res.hasNext());
          assertEquals("s1", res.next().getString(0));
          res = s1.sql("SHOW CREATE TABLE testpooledsessionstmps1").execute();
          assertTrue(res.hasNext());
          assertEquals("testpooledsessionstmps1", res.next().getString(0));
  
          s0.close();
          s1.close();
  
          Session s0_new = cli0.getSession();
          Session s1_new = cli0.getSession();
  
          assertEquals(id0, ((SessionImpl) s0_new).getSession().getServerSession().getCapabilities().getThreadId());
          assertEquals(id1, ((SessionImpl) s1_new).getSession().getServerSession().getCapabilities().getThreadId());
  
          res = s0_new.sql("SELECT @a as a").execute();
          assertTrue(res.hasNext());
          assertNull(res.next().getString(0));
  
          assertThrows(XProtocolError.class, ".*testpooledsessionstmps0' doesn't exist", new Callable<Void>() {
              public Void call() throws Exception {
                  s0_new.sql("SHOW CREATE TABLE testpooledsessionstmps0").execute();
                  return null;
              }
          });
  
          res = s1_new.sql("SELECT @a as a").execute();
          assertTrue(res.hasNext());
          assertNull(res.next().getString(0));
  
          assertThrows(XProtocolError.class, ".*testpooledsessionstmps1' doesn't exist", new Callable<Void>() {
              public Void call() throws Exception {
                  s1_new.sql("SHOW CREATE TABLE testpooledsessionstmps1").execute();
                  return null;
              }
          });
  
          cli0.close();
      }
  
      private void testPooledSessions_checkClientProperties(Client cli, int maxSize, int maxIdleTime, int queueTimeout) throws Exception {
          Field f = ClientImpl.class.getDeclaredField("maxSize");
          f.setAccessible(true);
          assertEquals(maxSize, f.get(cli));
  
          f = ClientImpl.class.getDeclaredField("maxIdleTime");
          f.setAccessible(true);
          assertEquals(maxIdleTime, f.get(cli));
  
          f = ClientImpl.class.getDeclaredField("queueTimeout");
          f.setAccessible(true);
          assertEquals(queueTimeout, f.get(cli));
      }
  
      private <EX extends Throwable> void testPooledSessions_assertFailureTimeout(Client cli, int expLowLimit, int expUpLimit, Class<EX> throwable,
              String message) {
          long begin = System.currentTimeMillis();
          assertThrows(throwable, message, () -> cli.getSession());
          long end = System.currentTimeMillis() - begin;
          assertTrue(end >= expLowLimit && end < expUpLimit, "Expected: " + expLowLimit + ".." + expUpLimit + ". Got " + end);
      }
  
      @Test
      public void testBug28616573() throws Exception {
          RowResult res = this.session.sql(
                  "select @@global.mysqlx_max_connections, VARIABLE_VALUE FROM performance_schema.global_status WHERE VARIABLE_NAME='Mysqlx_worker_threads_active'")
                  .execute();
          Row r = res.next();
          int mysqlxMaxConnections = r.getInt(0);
          int mysqlWorkerThreadsActive = Integer.parseInt(r.getString(1));
          this.session.sql("SET @@global.mysqlx_max_connections=" + (mysqlWorkerThreadsActive + 2)).execute(); // allow only 2 additional connections
  
          Properties props = new Properties();
          props.setProperty(ClientProperty.POOLING_ENABLED.getKeyName(), "true");
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "2");
          props.setProperty(ClientProperty.POOLING_MAX_IDLE_TIME.getKeyName(), "2000");
          props.setProperty(ClientProperty.POOLING_QUEUE_TIMEOUT.getKeyName(), "2000");
  
          try {
              ClientFactory cf = new ClientFactory();
              Client cli1 = cf.getClient(this.baseUrl, props);
              Client cli2 = cf.getClient(this.baseUrl, props);
  
              Session sess1 = cli1.getSession(); // new connection #1
              sess1.sql("SELECT 1").execute();
              sess1.close();
              sess1 = cli1.getSession(); // reuse connection #1
              Session sess2 = cli1.getSession(); // new connection #2
              sess2.sql("SELECT 1").execute();
  
              assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", cli2::getSession);
  
              cli1.close();
              cli2.close();
  
          } finally {
              this.session.sql("SET @@global.mysqlx_max_connections=" + mysqlxMaxConnections).execute();
          }
      }
  
      @Test
      public void testBug28606708() throws Exception {
          assumeTrue(isServerRunningOnWindows() && isMysqlRunningLocally(),
                  "This test can run only when client and server are running on the same Windows host.");
  
          for (String path : new String[] { null, "\\\\.\\pipe\\MySQL80" }) {
              String url = this.baseUrl + makeParam(PropertyKey.socketFactory, "com.mysql.cj.protocol.NamedPipeSocketFactory");
              if (path != null) {
                  url += makeParam(PropertyKey.PATH, path);
              }
              try {
                  this.fact.getSession(url);
                  fail("The named-pipe connection attempt must fail with " + (path != null ? path : "default") + " path.");
              } catch (Exception e) {
                  Throwable cause = e.getCause();
                  if (cause != null) {
                      if (cause instanceof CJCommunicationsException && cause.getCause() != null && cause.getCause() instanceof FileNotFoundException
                              && ((path == null ? "\\\\.\\pipe\\MySQL" : path) + " (The system cannot find the file specified)")
                                      .equals(cause.getCause().getMessage())) {
                          continue;
                      } else if (cause instanceof XProtocolError
                              && "ASSERTION FAILED: Unknown message type: 10 (server messages mapping: null)".equals(cause.getMessage())) {
                          // if named pipes are enabled on server then we expect this error because the pipe is bound to legacy protocol
                          continue;
                      }
                  }
                  throw e;
              }
          }
      }
  
      @Test
      public void testSessionAttributes() throws Exception {
          assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.16")), "MySQL 8.0.16+ is required to run this test.");
  
          ClientFactory cf = new ClientFactory();
          Map<String, String> userAttributes = new HashMap<>();
  
          // TSFR1/TSFR2/TSFR3 Create a Session with xdevapi.connection-attributes in the connection string, verify that
          // the session is successfully established. Create a Session using a connection string containing the properties
          // as listed below, verify that the session is successfully established and the server contains the defined session attributes.
          //    xdevapi.connection-attributes=[key1=value1]
          //    xdevapi.connection-attributes=[key1=value1,key2=value2]
          //    xdevapi.connection-attributes=key1=value1
          //    xdevapi.connection-attributes=key1=value1,key2=value2
          userAttributes.clear();
          userAttributes.put("key1", "value1");
  
          String baseUrlLocal = this.baseUrl + (this.baseUrl.contains("?") ? "&" : "?");
  
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1]", true), userAttributes);
  
          userAttributes.put("key2", "value2");
  
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,key2=value2", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1,key2=value2]", true),
                  userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,key2=value2", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1,key2=value2]", true), userAttributes);
  
          // TSFR4/TSFR5/TSFR6 Create a Session without xdevapi.connection-attributes in the connection string, verify that all predefined attributes
          // exist and contain the correct values. Verify that only connection attributes starting with "_" were set for current session.
          userAttributes.clear();
          testSessionAttributes_checkSession(this.baseUrl, userAttributes);
          testSessionAttributes_checkClient(this.baseUrl, userAttributes);
  
          // TSFR7 Create a Session using a connection string containing the properties as listed below, verify that a WrongArgumentException exception is thrown
          // with the message Key names in "xdevapi.connection-attributes" cannot start with "_".
          //    xdevapi.connection-attributes=[_key1=value1]
          //    xdevapi.connection-attributes=[key1=value1,_key2=value2]
          //    xdevapi.connection-attributes=_key1=value1
          //    xdevapi.connection-attributes=key1=value1,_key2=value2
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[_key1=value1]", true));
              return null;
          });
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              Client cli1 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[_key1=value1]", true), new Properties());
              cli1.getSession();
              return null;
          });
  
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1,_key2=value2]", true));
              return null;
          });
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              Client cli1 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=value1,_key2=value2]", true), new Properties());
              cli1.getSession();
              return null;
          });
  
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "_key1=value1", true));
              return null;
          });
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              Client cli1 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "_key1=value1", true), new Properties());
              cli1.getSession();
              return null;
          });
  
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,_key2=value2", true));
              return null;
          });
          assertThrows(WrongArgumentException.class, "Key names in \"xdevapi.connection-attributes\" cannot start with \"_\".", () -> {
              Client cli1 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,_key2=value2", true), new Properties());
              cli1.getSession();
              return null;
          });
  
          // TSFR9 Create a Session using a connection string containing the properties as listed below, verify
          // that the user-defined connection attribute key1 has an empty value.
          //    xdevapi.connection-attributes=[key1]
          //    xdevapi.connection-attributes=[key1,key2=value2]
          //    xdevapi.connection-attributes=[key1=]
          //    xdevapi.connection-attributes=[key1=,key2=value2]
          //    xdevapi.connection-attributes=key1
          //    xdevapi.connection-attributes=key1,key2=value2
          //    xdevapi.connection-attributes=key1=
          //    xdevapi.connection-attributes=key1=,key2=value2
          userAttributes.clear();
          userAttributes.put("key1", "");
  
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1]", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=]", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=", true), userAttributes);
  
          userAttributes.put("key2", "value2");
  
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1,key2=value2]", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=,key2=value2]", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1,key2=value2", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=,key2=value2", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1,key2=value2]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[key1=,key2=value2]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1,key2=value2", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=,key2=value2", true), userAttributes);
  
          // TSFR10 Create a Session with xdevapi.connection-attributes=false in the connection string, verify
          // that no connection attribute was set for current session.
          Session s10 = this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "false", true));
          SqlResult res = s10.sql("SELECT * FROM performance_schema.session_connect_attrs WHERE processlist_id = CONNECTION_ID()").execute();
          assertFalse(res.hasNext(), "Expected no connection attributes.");
          s10.close();
  
          Client c10 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "false", true), new Properties());
          s10 = c10.getSession();
          res = s10.sql("SELECT * FROM performance_schema.session_connect_attrs WHERE processlist_id = CONNECTION_ID()").execute();
          assertFalse(res.hasNext(), "Expected no connection attributes.");
          s10.close();
          c10.close();
  
          // TSFR11 Create a Session using a connection string containing the properties as listed below, verify
          // that only the client-defined connection attributes were set for the current session.
          //    xdevapi.connection-attributes
          //    xdevapi.connection-attributes=
          //    xdevapi.connection-attributes=true
          //    xdevapi.connection-attributes=[]
          userAttributes.clear();
  
          testSessionAttributes_checkSession(baseUrlLocal + PropertyKey.xdevapiConnectionAttributes.getKeyName(), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + PropertyKey.xdevapiConnectionAttributes.getKeyName() + "=", userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "true", true), userAttributes);
          testSessionAttributes_checkSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[]", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + PropertyKey.xdevapiConnectionAttributes.getKeyName(), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + PropertyKey.xdevapiConnectionAttributes.getKeyName() + "=", userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "true", true), userAttributes);
          testSessionAttributes_checkClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "[]", true), userAttributes);
  
          // TSFR13 Create a Session with xdevapi.connection-attributes=[key1=value1,key1=value2] in the connection string, verify that
          // a WrongArgumentException exception is thrown with the message Duplicate key "key1" used in "xdevapi.connection-attributes".
          assertThrows(WrongArgumentException.class, "Duplicate key \"key1\" used in \"xdevapi.connection-attributes\".", () -> {
              this.fact.getSession(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,key1=value2", true));
              return null;
          });
          assertThrows(WrongArgumentException.class, "Duplicate key \"key1\" used in \"xdevapi.connection-attributes\".", () -> {
              Client cli1 = cf.getClient(baseUrlLocal + makeParam(PropertyKey.xdevapiConnectionAttributes, "key1=value1,key1=value2", true), new Properties());
              cli1.getSession();
              return null;
          });
      }
  
      private void testSessionAttributes_checkSession(String url, Map<String, String> userAttributes) {
          Session s = this.fact.getSession(url);
          try {
              testSessionAttributes_checkSession(s, userAttributes);
          } finally {
              s.close();
          }
      }
  
      private void testSessionAttributes_checkSession(Session s, Map<String, String> userAttributes) {
          Map<String, Integer> matchedCounts = new HashMap<>();
          Map<String, String> matchValues = new HashMap<>();
          matchValues.put("_platform", Constants.OS_ARCH);
          matchValues.put("_os", Constants.OS_NAME + "-" + Constants.OS_VERSION);
          matchValues.put("_client_name", Constants.CJ_NAME);
          matchValues.put("_client_version", Constants.CJ_VERSION);
          matchValues.put("_client_license", Constants.CJ_LICENSE);
          matchValues.put("_runtime_version", Constants.JVM_VERSION);
          matchValues.put("_runtime_vendor", Constants.JVM_VENDOR);
          matchValues.putAll(userAttributes);
  
          SqlResult res = s.sql("SELECT * FROM performance_schema.session_connect_attrs WHERE processlist_id = CONNECTION_ID()").execute();
          while (res.hasNext()) {
              Row r = res.next();
              String key = r.getString(1);
              String val = r.getString(2);
              assertTrue(matchValues.containsKey(key), "Unexpected connection attribute key:  " + key);
              Integer cnt = matchedCounts.get(key);
              matchedCounts.put(key, cnt == null ? 1 : cnt++);
  
              // when client sends an empty string as an attribute value the NULL value is stored to performance_schema.session_connect_attrs
              String expected = matchValues.get(key);
              if (expected.equals("")) {
                  expected = null;
              }
              assertEquals(expected, val);
          }
          for (String key : matchValues.keySet()) {
              assertTrue(matchedCounts.containsKey(key), "Incorrect number of entries for key \"" + key + "\": 0");
              assertTrue(matchedCounts.get(key) == 1, "Incorrect number of entries for key \"" + key + "\": " + matchedCounts.get(key));
          }
      }
  
      private void testSessionAttributes_checkClient(String url, Map<String, String> userAttributes) throws Exception {
          ClientFactory cf = new ClientFactory();
          Client c = cf.getClient(url, new Properties());
  
          try {
              Session s0 = c.getSession();
              testSessionAttributes_checkSession(s0, userAttributes);
              s0.close(); // return to pool
  
              // check that pooled session set the same attributes
              Session s1 = c.getSession(); // get it from pool
              s1.sql("SELECT 1").execute();
              assertNotEquals(s0, s1);
              Field fProtocol = CoreSession.class.getDeclaredField("protocol");
              fProtocol.setAccessible(true);
              assertEquals(fProtocol.get(((SessionImpl) s0).getSession()), fProtocol.get(((SessionImpl) s1).getSession()));
              testSessionAttributes_checkSession(s1, userAttributes);
          } finally {
              c.close();
          }
      }
  
      @Test
      public void testPreparedStatementsCleanup() {
          assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.14")), "MySQL 8.0.14+ is required to run this test.");
  
          try {
              // Prepare test data.
              this.schema.createCollection("testPrepStmtClean", true).add("{\"_id\":\"1\"}").execute();
  
              SessionFactory sf = new SessionFactory();
  
              /*
               * Test common usage.
               */
              Session testSession = sf.getSession(this.testProperties);
  
              int sessionThreadId = getThreadId(testSession);
              assertPreparedStatementsCount(sessionThreadId, 0, 1);
              assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);
  
              // Initialize several *Statement objects.
              FindStatement testFind1 = testSession.getDefaultSchema().getCollection("testPrepStmtClean").find();
              SelectStatement testSelect1 = testSession.getDefaultSchema().getCollectionAsTable("testPrepStmtClean").select("_id");
              FindStatement testFind2 = testSession.getDefaultSchema().getCollection("testPrepStmtClean").find();
              SelectStatement testSelect2 = testSession.getDefaultSchema().getCollectionAsTable("testPrepStmtClean").select("_id");
  
              // 1st execute -> don't prepare.
              testFind1.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testFind1, 0, -1);
              testSelect1.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testSelect1, 0, -1);
              testFind2.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testFind2, 0, -1);
              testSelect2.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
  
              assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);
  
              // 2nd execute -> prepare + execute.
              testFind1.execute();
              assertPreparedStatementsCountsAndId(testSession, 1, testFind1, 1, 1);
              testSelect1.execute();
              assertPreparedStatementsCountsAndId(testSession, 2, testSelect1, 2, 1);
              testFind2.execute();
              assertPreparedStatementsCountsAndId(testSession, 3, testFind2, 3, 1);
              testSelect2.execute();
              assertPreparedStatementsCountsAndId(testSession, 4, testSelect2, 4, 1);
  
              assertPreparedStatementsStatusCounts(testSession, 4, 4, 0);
              assertPreparedStatementsCount(sessionThreadId, 4, 1);
  
              /*
               * The following verifications are non-deterministic as System.gc() only hints the JVM to perform a garbage collection. This approach allows some
               * time for the JVM to execute the GC. In case of failure the repeats or wait times may have to be adjusted.
               * The test can be deleted entirely if no reasonable setup can be found.
               */
  
              // Nullify first statement.
              testFind1 = null;
              System.gc();
              int psCount, countdown = 10;
              do {
                  try {
                      Thread.sleep(100);
                  } catch (InterruptedException e) {
                  }
                  testSession.sql("SELECT 1").execute();
                  psCount = getPreparedStatementsCount(sessionThreadId);
              } while (psCount != 3 && --countdown > 0);
              assertPreparedStatementsStatusCounts(testSession, 4, 4, 1);
              assertPreparedStatementsCount(sessionThreadId, 3, 1);
  
              // Nullify second and third statements.
              testSelect1 = null;
              testFind2 = null;
              System.gc();
              countdown = 10;
              do {
                  try {
                      Thread.sleep(100);
                  } catch (InterruptedException e) {
                  }
                  testSession.sql("SELECT 1").execute();
                  psCount = getPreparedStatementsCount(sessionThreadId);
              } while (psCount != 1 && --countdown > 0);
              assertPreparedStatementsStatusCounts(testSession, 4, 4, 3);
              assertPreparedStatementsCount(sessionThreadId, 1, 1);
  
              // Nullify last statement.
              testSelect2 = null;
              System.gc();
              countdown = 10;
              do {
                  try {
                      Thread.sleep(100);
                  } catch (InterruptedException e) {
                  }
                  testSession.sql("SELECT 1").execute();
                  psCount = getPreparedStatementsCount(sessionThreadId);
              } while (psCount != 0 && --countdown > 0);
              assertPreparedStatementsStatusCounts(testSession, 4, 4, 4);
              assertPreparedStatementsCount(sessionThreadId, 0, 1);
  
              testSession.close();
              assertPreparedStatementsCount(sessionThreadId, 0, 1);
          } finally {
              this.schema.dropCollection("testPrepStmtClean");
          }
      }
  
      @Test
      public void testPreparedStatementsPooledConnections() {
          assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.14")), "MySQL 8.0.14+ is required to run this test.");
  
          Properties props = new Properties();
          props.setProperty(ClientProperty.POOLING_ENABLED.getKeyName(), "true");
          props.setProperty(ClientProperty.POOLING_MAX_SIZE.getKeyName(), "1");
  
          try {
              this.schema.createCollection("testPrepStmtPooling", true).add("{\"_id\":\"1\"}").execute();
  
              ClientFactory cf = new ClientFactory();
              Client testClient = cf.getClient(this.baseUrl, props);
  
              Session testSession = testClient.getSession();
              int sessionThreadId = getThreadId(testSession);
              assertPreparedStatementsCount(sessionThreadId, 0, 1);
  
              FindStatement testFind = testSession.getDefaultSchema().getCollection("testPrepStmtPooling").find();
  
              // 1st execute -> don't prepare.
              testFind.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testFind, 0, -1);
              assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);
  
              // 2nd execute -> prepare + execute.
              testFind.execute();
              assertPreparedStatementsCountsAndId(testSession, 1, testFind, 1, 1);
              assertPreparedStatementsStatusCounts(testSession, 1, 1, 0);
  
              assertPreparedStatementsCount(sessionThreadId, 1, 1);
  
              // Prepared statements won't live past closing the session, or returning it to the pool.
              testSession.close();
              assertPreparedStatementsCount(sessionThreadId, 0, 10);
  
              testSession = testClient.getSession();
              sessionThreadId = getThreadId(testSession);
              assertPreparedStatementsCount(sessionThreadId, 0, 1);
  
              // The underlying connection object in testFind is the same as the one returned from the pool to the new session.
              assertThrows(XProtocolError.class, "ERROR 5110 \\(HY000\\) Statement with ID=1 was not prepared", testFind::execute); // This exec attempt counts.
              if (mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.16"))) { // Mysqlx.Session.Reset doesn't clear PS counters.
                  assertPreparedStatementsStatusCounts(testSession, 1, 2, 0);
              } else {
                  assertPreparedStatementsStatusCounts(testSession, 0, 1, 0);
              }
  
              testFind = testSession.getDefaultSchema().getCollection("testPrepStmtPooling").find();
  
              // 1st execute -> don't prepare.
              testFind.execute();
              assertPreparedStatementsCountsAndId(testSession, 0, testFind, 0, -1);
              if (mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.16"))) { // Mysqlx.Session.Reset doesn't clear PS counters.
                  assertPreparedStatementsStatusCounts(testSession, 1, 2, 0);
              } else {
                  assertPreparedStatementsStatusCounts(testSession, 0, 1, 0);
              }
              // 2nd execute -> prepare + execute.
              testFind.execute();
              assertPreparedStatementsCountsAndId(testSession, 1, testFind, 1, 1);
              if (mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.16"))) { // Mysqlx.Session.Reset doesn't clear PS counters.
                  assertPreparedStatementsStatusCounts(testSession, 2, 3, 0);
              } else {
                  assertPreparedStatementsStatusCounts(testSession, 1, 2, 0);
              }
  
              assertPreparedStatementsCount(sessionThreadId, 1, 1);
  
              // Prepared statements won't live past closing the client and its sessions.
              testClient.close();
              assertPreparedStatementsCount(sessionThreadId, 0, 10);
  
              assertThrows(CJCommunicationsException.class, "Unable to write message", testFind::execute);
          } finally {
              this.schema.dropCollection("testPrepStmtPooling");
          }
      }
  
      @Test
      public void testBug23721537() throws Exception {
          try {
              sqlUpdate("drop table if exists testBug23721537");
              sqlUpdate("create table testBug23721537 (id int, name varchar(20) not null)");
              sqlUpdate("insert into testBug23721537 values (0, 'a')");
  
              this.session.sql("drop procedure if exists newproc").execute();
              this.session.sql(
                      "create procedure newproc (in p1 int,in p2 char(20)) begin select 1; update testBug23721537 set name='b' where id=0; select 2; select 3; end;")
                      .execute();
  
              /* sync execution */
              SqlResult res1 = this.session.sql("call newproc(?,?)").bind(10).bind("X").execute();
  
              assertTrue(res1 instanceof SqlMultiResult);
              assertTrue(res1.hasData());
              assertTrue(res1.hasNext());
              Row r = res1.next();
              assertEquals(1, r.getInt(0));
              assertFalse(res1.hasNext());
  
              SqlResult res2 = this.session.sql("select 10").execute(); // res1 should finish streaming here
  
              assertTrue(res1.nextResult());
              assertTrue(res1.hasData());
              assertTrue(res1.hasNext());
              r = res1.next();
              assertEquals(2, r.getInt(0));
              assertFalse(res1.hasNext());
  
              assertTrue(res1.nextResult());
              assertTrue(res1.hasData());
              assertTrue(res1.hasNext());
              r = res1.next();
              assertEquals(3, r.getInt(0));
              assertFalse(res1.hasNext());
  
              assertFalse(res1.nextResult());
  
              //
              assertTrue(res2.hasData());
              assertTrue(res2.hasNext());
              r = res2.next();
              assertEquals(10, r.getInt(0));
              assertFalse(res2.hasNext());
              assertFalse(res2.nextResult());
  
              /* async execution */
              res1 = this.session.sql("call newproc(?,?)").bind(10).bind("X").executeAsync().get();
  
              assertTrue(res1.hasData());
              r = res1.next();
              assertEquals(1, r.getInt(0));
              assertFalse(res1.hasNext());
  
              res2 = this.session.sql("select 10").executeAsync().get(); // res1 should finish streaming here
  
              assertTrue(res1.nextResult());
              assertTrue(res1.hasData());
              assertTrue(res1.hasNext());
              r = res1.next();
              assertEquals(2, r.getInt(0));
              assertFalse(res1.hasNext());
  
              assertTrue(res1.nextResult());
              assertTrue(res1.hasData());
              assertTrue(res1.hasNext());
              r = res1.next();
              assertEquals(3, r.getInt(0));
              assertFalse(res1.hasNext());
  
              assertFalse(res1.nextResult());
  
              //
              assertTrue(res2.hasData());
              assertTrue(res2.hasNext());
              r = res2.next();
              assertEquals(10, r.getInt(0));
              assertFalse(res2.hasNext());
              assertFalse(res2.nextResult());
  
          } finally {
              sqlUpdate("drop table if exists testBug23721537");
              this.session.sql("drop procedure if exists newproc").execute();
          }
      }
  
      @Test
      public void basicSessionFailoverRandomSort() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> h + ":" + getTestPort()).collect(Collectors.joining(","));
          String testUri = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(), getTestPassword() == null ? "" : getTestPassword(),
                  testHosts, getTestDatabase());
  
          Set<String> downHosts = new HashSet<>();
          for (int i = 1; i <= 5; i++) {
              Session testSession = this.fact.getSession(testUri);
              assertTrue(UnreliableSocketFactory.isConnected());
              testSession.close();
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              for (String h : downHosts) {
                  assertNotEquals(h, lastHost);
              }
              downHosts.add(lastHost);
              UnreliableSocketFactory.downHost(lastHost.substring(1));
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              this.fact.getSession(testUri);
              return null;
          });
  
          UnreliableSocketFactory.dontDownHost("host3");
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession = this.fact.getSession(testUri);
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession.close();
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host3"), UnreliableSocketFactory.getHostFromLastConnection());
      }
  
      @Test
      public void basicSessionFailoverByPriorities() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> "(address=" + h + ":" + getTestPort() + ",priority=%d)").collect(Collectors.joining(","));
          String testUriPatternPriorities = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(),
                  getTestPassword() == null ? "" : getTestPassword(), testHosts, getTestDatabase());
          String testUri = String.format(testUriPatternPriorities, 60, 80, 100, 20, 40);
          int[] hostsOrder = new int[] { 3, 2, 1, 5, 4 };
  
          for (int i = 0; i < hostsOrder.length; i++) {
              int h = hostsOrder[i];
  
              Session testSession = this.fact.getSession(testUri);
              assertTrue(UnreliableSocketFactory.isConnected());
              testSession.close();
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + h), lastHost);
              List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
              for (int a = 0; a < i; a++) {
                  assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[a]), connectionAttempts.get(a));
              }
              UnreliableSocketFactory.downHost("host" + h);
              UnreliableSocketFactory.flushConnectionAttempts();
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              this.fact.getSession(testUri);
              return null;
          });
  
          UnreliableSocketFactory.dontDownHost("host" + hostsOrder[1]);
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession = this.fact.getSession(testUri);
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession.close();
          List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          assertEquals(2, connectionAttempts.size());
          assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[0]), connectionAttempts.get(0));
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + hostsOrder[1]), connectionAttempts.get(1));
      }
  
      @Test
      public void pooledSessionFailoverRandomSortAndPooling() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> h + ":" + getTestPort()).collect(Collectors.joining(","));
          String testUri = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(), getTestPassword() == null ? "" : getTestPassword(),
                  testHosts, getTestDatabase());
  
          final ClientFactory cf = new ClientFactory();
          Client client = cf.getClient(testUri, "{\"pooling\" : {\"enabled\" : true, \"maxSize\" : 10} }");
  
          Set<String> downHosts = new HashSet<>();
          for (int i = 1; i <= 5; i++) {
              Session testSession = client.getSession();
              assertTrue(UnreliableSocketFactory.isConnected());
              testSession.close(); // Pool this connection.
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              for (String h : downHosts) {
                  assertNotEquals(h, lastHost);
              }
              downHosts.add(lastHost);
              UnreliableSocketFactory.downHost(lastHost.substring(1));
              UnreliableSocketFactory.flushConnectionAttempts();
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              client.getSession();
              return null;
          });
  
          UnreliableSocketFactory.dontDownHost("host3");
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession = client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession.close();
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host3"), UnreliableSocketFactory.getHostFromLastConnection());
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession1 = client.getSession(); // Pick previous connection from the pool. Doesn't count as new connections.
          assertEquals(0, UnreliableSocketFactory.getHostsFromAllConnections().size());
          Session testSession2 = client.getSession(); // Create a new connection.
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession1.close();
          testSession2.close();
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host3"), UnreliableSocketFactory.getHostFromLastConnection());
  
          client.close();
      }
  
      @Test
      public void pooledSessionFailoverRandomSortAndNoPooling() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> h + ":" + getTestPort()).collect(Collectors.joining(","));
          String testUri = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(), getTestPassword() == null ? "" : getTestPassword(),
                  testHosts, getTestDatabase());
  
          final ClientFactory cf = new ClientFactory();
          Client client = cf.getClient(testUri, "{\"pooling\" : {\"enabled\" : true, \"maxSize\" : 10} }");
  
          Set<String> downHosts = new HashSet<>();
          for (int i = 1; i <= 5; i++) {
              client.getSession(); // Don't pool this connection.
              assertTrue(UnreliableSocketFactory.isConnected());
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              for (String h : downHosts) {
                  assertNotEquals(h, lastHost);
              }
              downHosts.add(lastHost);
              UnreliableSocketFactory.downHost(lastHost.substring(1));
              UnreliableSocketFactory.flushConnectionAttempts();
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              client.getSession();
              return null;
          });
  
          UnreliableSocketFactory.dontDownHost("host3");
  
          UnreliableSocketFactory.flushConnectionAttempts();
          client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host3"), UnreliableSocketFactory.getHostFromLastConnection());
  
          UnreliableSocketFactory.flushConnectionAttempts();
          client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host3"), UnreliableSocketFactory.getHostFromLastConnection());
  
          client.close();
      }
  
      @Test
      public void pooledSessionFailoverByPrioritiesAndPooling() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> "(address=" + h + ":" + getTestPort() + ",priority=%d)").collect(Collectors.joining(","));
          String testUriPatternPriorities = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(),
                  getTestPassword() == null ? "" : getTestPassword(), testHosts, getTestDatabase());
          String testUri = String.format(testUriPatternPriorities, 60, 80, 100, 20, 40);
          int[] hostsOrder = new int[] { 3, 2, 1, 5, 4 };
  
          final ClientFactory cf = new ClientFactory();
          Client client = cf.getClient(testUri, "{\"pooling\" : {\"enabled\" : true, \"maxSize\" : 10} }");
  
          for (int i = 0; i < hostsOrder.length; i++) {
              int h = hostsOrder[i];
  
              Session testSession = client.getSession();
              assertTrue(UnreliableSocketFactory.isConnected());
              testSession.close(); // Pool this connection.
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + h), lastHost);
              List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
              assertEquals(i == 0 ? 1 : 2, connectionAttempts.size());
              for (int a = 0; a < connectionAttempts.size() - 1; a++) {
                  assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[i - 1 + a]), connectionAttempts.get(a));
              }
              UnreliableSocketFactory.downHost("host" + h);
              UnreliableSocketFactory.flushConnectionAttempts();
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              client.getSession();
              return null;
          });
          // Final connection tried the last known to be good host (host4) and then the remaining hosts by their priority
          List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          for (int i = 0; i < hostsOrder.length; i++) {
              assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[(i + 4) % 5]), connectionAttempts.get(i));
          }
  
          UnreliableSocketFactory.dontDownHost("host" + hostsOrder[1]);
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession = client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession.close();
          connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          assertEquals(2, connectionAttempts.size());
          assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[0]), connectionAttempts.get(0));
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + hostsOrder[1]), connectionAttempts.get(1));
  
          UnreliableSocketFactory.flushConnectionAttempts();
          Session testSession1 = client.getSession(); // Pick previous connection from the pool. Doesn't count as new connections.
          assertEquals(0, UnreliableSocketFactory.getHostsFromAllConnections().size());
          Session testSession2 = client.getSession(); // Create a new connection.
          assertTrue(UnreliableSocketFactory.isConnected());
          testSession1.close();
          testSession2.close();
          connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          assertEquals(1, connectionAttempts.size());
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + hostsOrder[1]), connectionAttempts.get(0));
  
          client.close();
      }
  
      @Test
      public void pooledSessionFailoverByPrioritiesAndNoPooling() {
          UnreliableSocketFactory.flushAllStaticData();
  
          final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + PropertyKey.xdevapiConnectTimeout.getKeyName() + "=100&"
                  + PropertyKey.socketFactory.getKeyName() + "=" + UnreliableSocketFactory.class.getName();
          String testHosts = IntStream.range(1, 6).mapToObj(i -> "host" + i).peek(h -> UnreliableSocketFactory.mapHost(h, getTestHost()))
                  .map(h -> "(address=" + h + ":" + getTestPort() + ",priority=%d)").collect(Collectors.joining(","));
          String testUriPatternPriorities = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(),
                  getTestPassword() == null ? "" : getTestPassword(), testHosts, getTestDatabase());
          String testUri = String.format(testUriPatternPriorities, 60, 80, 100, 20, 40);
          int[] hostsOrder = new int[] { 3, 2, 1, 5, 4 };
  
          final ClientFactory cf = new ClientFactory();
          Client client = cf.getClient(testUri, "{\"pooling\" : {\"enabled\" : true, \"maxSize\" : 10} }");
  
          for (int i = 0; i < hostsOrder.length; i++) {
              int h = hostsOrder[i];
  
              client.getSession(); // Don't pool this connection.
              assertTrue(UnreliableSocketFactory.isConnected());
  
              String lastHost = UnreliableSocketFactory.getHostFromLastConnection();
              assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + h), lastHost);
              List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
              assertEquals(i == 0 ? 1 : 2, connectionAttempts.size());
              for (int a = 0; a < connectionAttempts.size() - 1; a++) {
                  assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[i - 1 + a]), connectionAttempts.get(a));
              }
              UnreliableSocketFactory.downHost("host" + h);
              UnreliableSocketFactory.flushConnectionAttempts();
          }
  
          // None of the hosts is available by now.
          assertThrows(CJCommunicationsException.class, "Unable to connect to any of the target hosts\\.", () -> {
              client.getSession();
              return null;
          });
          // Final connection tried the last known to be good host (host4) and then the remaining hosts by their priority
          List<String> connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          for (int i = 0; i < hostsOrder.length; i++) {
              assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[(i + 4) % 5]), connectionAttempts.get(i));
          }
  
          UnreliableSocketFactory.dontDownHost("host" + hostsOrder[1]);
  
          UnreliableSocketFactory.flushConnectionAttempts();
          client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          assertEquals(2, connectionAttempts.size());
          assertEquals(UnreliableSocketFactory.getHostFailedStatus("host" + hostsOrder[0]), connectionAttempts.get(0));
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + hostsOrder[1]), connectionAttempts.get(1));
  
          UnreliableSocketFactory.flushConnectionAttempts();
          client.getSession();
          assertTrue(UnreliableSocketFactory.isConnected());
          connectionAttempts = UnreliableSocketFactory.getHostsFromAllConnections();
          assertEquals(1, connectionAttempts.size());
          assertEquals(UnreliableSocketFactory.getHostConnectedStatus("host" + hostsOrder[1]), connectionAttempts.get(0));
  
          client.close();
      }
  
      @Test
      public void testConnectionCloseNotification() throws Exception {
          String shutdownMessage = "Server shutdown in progress";
          String ioReadErrorMessage = "IO Read error: read_timeout exceeded";
          String sessionWasKilledMessage = "Session was killed";
  
          // create notice message buffers
          Frame.Builder shutdownNotice = Frame.newBuilder().setScope(Frame.Scope.GLOBAL).setType(Frame.Type.WARNING_VALUE)
                  .setPayload(com.mysql.cj.x.protobuf.MysqlxNotice.Warning.newBuilder().setCode(MysqlErrorNumbers.ER_SERVER_SHUTDOWN).setMsg(shutdownMessage)
                          .build().toByteString());
          Frame.Builder ioReadErrorNotice = Frame.newBuilder().setScope(Frame.Scope.GLOBAL).setType(Frame.Type.WARNING_VALUE)
                  .setPayload(com.mysql.cj.x.protobuf.MysqlxNotice.Warning.newBuilder().setCode(MysqlErrorNumbers.ER_IO_READ_ERROR).setMsg(ioReadErrorMessage)
                          .build().toByteString());
          Frame.Builder sessionWasKilledNotice = Frame.newBuilder().setScope(Frame.Scope.GLOBAL).setType(Frame.Type.WARNING_VALUE)
                  .setPayload(com.mysql.cj.x.protobuf.MysqlxNotice.Warning.newBuilder().setCode(MysqlErrorNumbers.ER_SESSION_WAS_KILLED)
                          .setMsg(sessionWasKilledMessage).build().toByteString());
  
          byte[] shutdownNoticeBytes = makeNoticeBytes(shutdownNotice.build());
          byte[] ioReadErrorNoticeBytes = makeNoticeBytes(ioReadErrorNotice.build());
          byte[] sessionWasKilledNoticeBytes = makeNoticeBytes(sessionWasKilledNotice.build());
  
          InjectedSocketFactory.flushAllStaticData();
  
          String url = this.baseUrl + (this.baseUrl.contains("?") ? "" : "?")
                  + makeParam(PropertyKey.socketFactory, InjectedSocketFactory.class.getName(), !this.baseUrl.contains("?") || this.baseUrl.endsWith("?"))
                  + makeParam(PropertyKey.xdevapiSslMode, XdevapiSslMode.DISABLED.toString())
                  + makeParam(PropertyKey.xdevapiCompression, Compression.DISABLED.toString())
                  // to allow injection between result rows
                  + makeParam(PropertyKey.useReadAheadInput, "false");
  
          // ER_SERVER_SHUTDOWN
  
          // 1.1. Shutdown during a sync query execution.
          Session sess11 = this.fact.getSession(url);
          SocketFactory sf = ((SessionImpl) sess11).getSession().getProtocol().getSocketConnection().getSocketFactory();
          assertTrue(InjectedSocketFactory.class.isAssignableFrom(sf.getClass()));
  
          InjectedSocketFactory.injectedBuffer = shutdownNoticeBytes;
          assertThrows(CJCommunicationsException.class, shutdownMessage, () -> sess11.sql("SELECT 1").execute());
  
          // 1.2. Shutdown during a sync result set consuming
          Session sess12 = this.fact.getSession(url);
          SqlResult res0 = sess12.sql("SELECT 1").execute();
          assertTrue(res0.hasNext());
          Row r = res0.next();
          assertEquals("1", r.getString(0));
          InjectedSocketFactory.injectedBuffer = shutdownNoticeBytes;
          assertThrows(CJCommunicationsException.class, shutdownMessage, () -> res0.next());
  
          // 1.3. Shutdown during an async query execution.
          Session sess13 = this.fact.getSession(url);
          InjectedSocketFactory.injectedBuffer = shutdownNoticeBytes;
          assertThrows(ExecutionException.class, CJCommunicationsException.class.getName() + ": " + shutdownMessage,
                  () -> sess13.sql("SELECT 1").executeAsync().get());
  
          // 1.4. Shutdown during an async result set consuming
          Session sess14 = this.fact.getSession(url);
          SqlResult res1 = sess14.sql("SELECT 1").executeAsync().get();
          InjectedSocketFactory.injectedBuffer = shutdownNoticeBytes;
          assertTrue(res1.hasNext());
          r = res1.next();
          assertEquals("1", r.getString(0));
          assertFalse(res1.hasNext());
          assertThrows(CJCommunicationsException.class, shutdownMessage, () -> sess14.getSchemas());
  
          // ER_IO_READ_ERROR
  
          // 2.1. ER_IO_READ_ERROR during a sync query execution.
          Session sess21 = this.fact.getSession(url);
          InjectedSocketFactory.injectedBuffer = ioReadErrorNoticeBytes;
          assertThrows(CJCommunicationsException.class, ioReadErrorMessage, () -> sess21.sql("SELECT 1").execute());
  
          // 2.2. ER_IO_READ_ERROR during a sync result set consuming
          Session sess22 = this.fact.getSession(url);
          SqlResult res2 = sess22.sql("SELECT 1").execute();
          assertTrue(res2.hasNext());
          r = res2.next();
          assertEquals("1", r.getString(0));
          InjectedSocketFactory.injectedBuffer = ioReadErrorNoticeBytes;
          assertThrows(CJCommunicationsException.class, ioReadErrorMessage, () -> res2.next());
  
          // 2.3. ER_IO_READ_ERROR during an async query execution.
          Session sess23 = this.fact.getSession(url);
          InjectedSocketFactory.injectedBuffer = ioReadErrorNoticeBytes;
          assertThrows(ExecutionException.class, CJCommunicationsException.class.getName() + ": " + ioReadErrorMessage,
                  () -> sess23.sql("SELECT 1").executeAsync().get());
  
          // 2.4. ER_IO_READ_ERROR during an async result set consuming
          Session sess24 = this.fact.getSession(url);
          res1 = sess24.sql("SELECT 1").executeAsync().get();
          InjectedSocketFactory.injectedBuffer = ioReadErrorNoticeBytes;
          assertTrue(res1.hasNext());
          r = res1.next();
          assertEquals("1", r.getString(0));
          assertFalse(res1.hasNext());
          assertThrows(CJCommunicationsException.class, ioReadErrorMessage, () -> sess24.getSchemas());
  
          // ER_SESSION_WAS_KILLED
  
          // 3.1. ER_SESSION_WAS_KILLED during a sync query execution.
          Session sess31 = this.fact.getSession(url);
          InjectedSocketFactory.injectedBuffer = sessionWasKilledNoticeBytes;
          assertThrows(CJCommunicationsException.class, sessionWasKilledMessage, () -> sess31.sql("SELECT 1").execute());
  
          // 3.2. ER_SESSION_WAS_KILLED during a sync result set consuming
          Session sess32 = this.fact.getSession(url);
          SqlResult res3 = sess32.sql("SELECT 1").execute();
          assertTrue(res3.hasNext());
          r = res3.next();
          assertEquals("1", r.getString(0));
          InjectedSocketFactory.injectedBuffer = sessionWasKilledNoticeBytes;
          assertThrows(CJCommunicationsException.class, sessionWasKilledMessage, () -> res3.next());
  
          // 3.3. ER_SESSION_WAS_KILLED during an async query execution.
          Session sess33 = this.fact.getSession(url);
          InjectedSocketFactory.injectedBuffer = sessionWasKilledNoticeBytes;
          assertThrows(ExecutionException.class, CJCommunicationsException.class.getName() + ": " + sessionWasKilledMessage,
                  () -> sess33.sql("SELECT 1").executeAsync().get());
  
          // 3.4. ER_SESSION_WAS_KILLED during an async result set consuming
          Session sess34 = this.fact.getSession(url);
          res1 = sess34.sql("SELECT 1").executeAsync().get();
          InjectedSocketFactory.injectedBuffer = sessionWasKilledNoticeBytes;
          assertTrue(res1.hasNext());
          r = res1.next();
          assertEquals("1", r.getString(0));
          assertFalse(res1.hasNext());
          assertThrows(CJCommunicationsException.class, sessionWasKilledMessage, () -> sess34.getSchemas());
  
          // TODO use a mock server instead of a real second server instance
          //        /*
          //         * With pooling.
          //         */
          //
          //        if (this.isSetForOpensslXTests) {
          //
          //            this.fact.getSession(this.baseOpensslUrl);
          //
          //            final String testUriPattern = "mysqlx://%s:%s@[%s]/%s?" + makeParam(PropertyKey.socketFactory, InjectedSocketFactory.class.getName())
          //                    + makeParam(PropertyKey.xdevapiSslMode, XdevapiSslMode.DISABLED.toString()) + makeParam(PropertyKey.allowPublicKeyRetrieval, "true")
          //                    + makeParam(PropertyKey.xdevapiCompression, Compression.DISABLED.toString())
          //                    // to allow injection between result rows
          //                    + makeParam(PropertyKey.useReadAheadInput, "false");
          //
          //            String testHosts = "(address=" + getTestHost() + ":" + getTestPort() + "),(address=" + getTestSslHost() + ":" + getTestSslPort() + ")";
          //            url = String.format(testUriPattern, getTestUser() == null ? "" : getTestUser(), getTestPassword() == null ? "" : getTestPassword(), testHosts,
          //                    getTestDatabase());
          //
          //            Field fProtocol = CoreSession.class.getDeclaredField("protocol");
          //            fProtocol.setAccessible(true);
          //
          //            Field idleProtocols = ClientImpl.class.getDeclaredField("idleProtocols");
          //            idleProtocols.setAccessible(true);
          //            Field activeProtocols = ClientImpl.class.getDeclaredField("activeProtocols");
          //            activeProtocols.setAccessible(true);
          //            Field nonPooledSessions = ClientImpl.class.getDeclaredField("nonPooledSessions");
          //            nonPooledSessions.setAccessible(true);
          //
          //            String host1 = getTestHost();
          //            String host2 = getTestSslHost();
          //            int port1 = getTestPort();
          //            int port2 = getTestSslPort();
          //
          //            final ClientFactory cf = new ClientFactory();
          //            Client cli0 = cf.getClient(url, "{\"pooling\": {\"enabled\": true}}");
          //
          //            InjectedSocketFactory.downHost(getTestSslHost() + ":" + getTestSslPort());
          //            Session s1_1 = cli0.getSession();
          //            Session s1_2 = cli0.getSession();
          //            Session s1_3 = cli0.getSession();
          //
          //            assertEquals(host1, ((SessionImpl) s1_1).getSession().getProcessHost());
          //            assertEquals(host1, ((SessionImpl) s1_2).getSession().getProcessHost());
          //            assertEquals(host1, ((SessionImpl) s1_3).getSession().getProcessHost());
          //            assertEquals(port1, ((SessionImpl) s1_1).getSession().getPort());
          //            assertEquals(port1, ((SessionImpl) s1_2).getSession().getPort());
          //            assertEquals(port1, ((SessionImpl) s1_3).getSession().getPort());
          //
          //            InjectedSocketFactory.dontDownHost(getTestSslHost() + ":" + getTestSslPort());
          //            InjectedSocketFactory.downHost(getTestHost() + ":" + getTestPort());
          //            Session s2_1 = cli0.getSession();
          //            Session s2_2 = cli0.getSession();
          //            Session s2_3 = cli0.getSession();
          //
          //            assertEquals(host2, ((SessionImpl) s2_1).getSession().getProcessHost());
          //            assertEquals(host2, ((SessionImpl) s2_2).getSession().getProcessHost());
          //            assertEquals(host2, ((SessionImpl) s2_3).getSession().getProcessHost());
          //            assertEquals(port2, ((SessionImpl) s2_1).getSession().getPort());
          //            assertEquals(port2, ((SessionImpl) s2_2).getSession().getPort());
          //            assertEquals(port2, ((SessionImpl) s2_3).getSession().getPort());
          //
          //            InjectedSocketFactory.dontDownHost(getTestHost() + ":" + getTestPort());
          //            InjectedSocketFactory.downHost(getTestSslHost() + ":" + getTestSslPort());
          //            Session s1 = cli0.getSession();
          //
          //            assertEquals(host1, ((SessionImpl) s1).getSession().getProcessHost());
          //            assertEquals(port1, ((SessionImpl) s1).getSession().getPort());
          //
          //            s1_1.close();
          //            s1_2.close();
          //            s1_3.close();
          //            s2_1.close();
          //            s2_2.close();
          //            s2_3.close();
          //
          //            assertEquals(1, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(6, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //
          //            // ER_IO_READ_ERROR
          //
          //            InjectedSocketFactory.injectedBuffer = ioReadErrorNoticeBytes;
          //            assertThrows(CJCommunicationsException.class, ioReadErrorMessage, () -> s1.sql("SELECT 1").execute());
          //
          //            assertEquals(0, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(6, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //            assertThrows(CJCommunicationsException.class, "Unable to write message", () -> s1.sql("SELECT 1").execute());
          //
          //            // ER_SESSION_WAS_KILLED
          //
          //            Session s2 = cli0.getSession();
          //            assertEquals(host1, ((SessionImpl) s2).getSession().getProcessHost());
          //            assertEquals(port1, ((SessionImpl) s2).getSession().getPort());
          //            assertEquals(1, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(5, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //
          //            InjectedSocketFactory.injectedBuffer = sessionWasKilledNoticeBytes;
          //            assertThrows(CJCommunicationsException.class, sessionWasKilledMessage, () -> s2.sql("SELECT 1").execute());
          //
          //            assertEquals(0, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(5, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //            assertThrows(CJCommunicationsException.class, "Unable to write message", () -> s2.sql("SELECT 1").execute());
          //
          //            // ER_SERVER_SHUTDOWN
          //
          //            Session s3 = cli0.getSession();
          //            assertEquals(host1, ((SessionImpl) s3).getSession().getProcessHost());
          //            assertEquals(port1, ((SessionImpl) s3).getSession().getPort());
          //            assertEquals(1, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(4, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //
          //            InjectedSocketFactory.injectedBuffer = shutdownNoticeBytes;
          //            assertThrows(CJCommunicationsException.class, shutdownMessage, () -> s3.sql("SELECT 1").execute());
          //
          //            assertEquals(0, ((Set<WeakReference<PooledXProtocol>>) activeProtocols.get(cli0)).size());
          //            assertEquals(3, ((BlockingQueue<PooledXProtocol>) idleProtocols.get(cli0)).size());
          //        }
      }
  
      private byte[] makeNoticeBytes(MessageLite msg) {
          int size = 1 + msg.getSerializedSize();
          byte[] nbuf = new byte[4 + size];
          System.arraycopy(ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(size).array(), 0, nbuf, 0, 4);
          nbuf[4] = (byte) ServerMessages.Type.NOTICE_VALUE;
          System.arraycopy(msg.toByteArray(), 0, nbuf, 5, size - 1);
          return nbuf;
      }
  
      /**
       * Tests fix for Bug#97730 (31699993), xdev api: ConcurrentModificationException at Session.close.
       * 
       * @throws Throwable
       */
      @Test
      public void testBug97730() throws Throwable {
          for (String pooling : new String[] { "false", "true" }) {
              Properties props = new Properties();
              props.setProperty(ClientProperty.POOLING_ENABLED.getKeyName(), pooling);
              ClientFactory cf = new ClientFactory();
              Client client = cf.getClient(this.baseUrl, props);
  
              List<Future<?>> futures = new ArrayList<>();
              ExecutorService executor = Executors.newFixedThreadPool(10);
              for (int i = 0; i < 1000; i++) {
                  futures.add(executor.submit(() -> {
                      client.getSession().close();
                  }));
              }
  
              Exception exception = null;
              for (Future<?> f : futures) {
                  try {
                      f.get();
                  } catch (ExecutionException e) {
                      exception = e;
                      break;
                  }
              }
              executor.shutdown();
              if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                  executor.shutdownNow();
              }
  
              if (exception != null) {
                  throw exception.getCause();
              }
          }
      }
  
      @Test
      public void testExecAsync() throws Exception {
          int i = 0;
          int NUMBER_OF_QUERIES = 5000;
          SqlResult sqlRes = null;
          Row r = null;
  
          CompletableFuture<SqlResult> asyncRes = null;
          long l1 = Long.MAX_VALUE, l2 = Long.MIN_VALUE;
          try {
              asyncRes = this.session.sql("drop Procedure if exists testExecAsyncProc").executeAsync();
              sqlRes = asyncRes.get();
  
              asyncRes = this.session.sql("create procedure testExecAsyncProc (in p1 int,in p2 int) begin select 1; select p1; select p2; end;").executeAsync();
              sqlRes = asyncRes.get();
  
              asyncRes = this.session.sql("drop table if exists testExecAsync").executeAsync();
              sqlRes = asyncRes.get();
  
              asyncRes = this.session.sql(
                      "create table testExecAsync (c1 int,c2 bigint,c3 double,c4 bool,c5 year,c6 float,c7 blob,c8 enum('xs','s','m','l','xl'),c9 set('a','b','c','d'))")
                      .executeAsync();
              sqlRes = asyncRes.get();
  
              asyncRes = this.session.sql("insert into testExecAsync values(1," + l1 + ",100.11,true,2000,100.101,'v1','xs',('a,b'))").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(1, sqlRes.getAffectedItemsCount());
  
              asyncRes = this.session.sql("insert into testExecAsync values(2," + (l1 - 1) + ",101.11,false,2001,101.101,'v2','s',('b,c')),(3," + (l2 - 2)
                      + ",102.11,true,2002,102.101,'v3','m',('a,b,c'))").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(2, sqlRes.getAffectedItemsCount());
  
              asyncRes = this.session.sql("insert into testExecAsync values(4," + (l1 - 4) + ",104.11,false,2004,104.101,'v4','s',('b,c,d')),(5," + (l2 - 5)
                      + ",105.11,true,2005,105.101,'v5','m',('a,c'))").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(2, sqlRes.getAffectedItemsCount());
  
              this.session.startTransaction();
  
              asyncRes = this.session.sql("insert into testExecAsync select * from testExecAsync").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(5, sqlRes.getAffectedItemsCount());
  
              asyncRes = this.session.sql(
                      "select c1 as 'col1',c2 as 'col2', c3 as 'col3',c4 as 'col4',c5 as 'col5', c6 as 'col6', c7 as 'col7', c8 as 'col8' , c9 as 'col9' from testExecAsync order by c1 asc")
                      .executeAsync();
              sqlRes = asyncRes.get();
              assertTrue(sqlRes.hasData());
  
              while (sqlRes.hasNext()) {
                  r = sqlRes.next();
                  //                System.out.println("col1 :" + r.getInt("col1"));
                  //                System.out.println("col2 :" + r.getLong("col2"));
                  //                System.out.println("col3 :" + r.getBigDecimal("col3"));
                  //                System.out.println("col4 :" + r.getBoolean("col4"));
                  //                System.out.println("col5 :" + r.getInt("col5"));
                  //                System.out.println("col6 :" + r.getDouble("col6"));
                  //                System.out.println("col7 :" + r.getString("col7"));
                  //                System.out.println("col8 :" + r.getString("col8"));
                  //                System.out.println("col9 :" + r.getString("col9"));
              }
  
              asyncRes = this.session.sql("update testExecAsync set c2=c2-1").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(10, sqlRes.getAffectedItemsCount());
  
              this.session.rollback();
              asyncRes = this.session.sql("create  unique index idx on  testExecAsync(c1)").executeAsync();
              sqlRes = asyncRes.get();
  
              asyncRes = this.session.sql("select count(*) from testExecAsync").executeAsync();
              sqlRes = asyncRes.get();
              r = sqlRes.next();
              assertEquals(5, r.getInt(0));
  
              asyncRes = this.session.sql("Delete from testExecAsync where c1=5 ").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(1, sqlRes.getAffectedItemsCount());
  
              /* WIth Bind */
              asyncRes = this.session.sql("insert into testExecAsync values  (?,?,?,?,?,?,?,?,?)").bind(6).bind(l1 - 6).bind(106.11).bind(true).bind(2006)
                      .bind(106.101).bind("v6").bind("xl").bind("a,a,a,a,a,a,a,a,a").executeAsync();
              sqlRes = asyncRes.get();
              assertEquals(1, sqlRes.getAffectedItemsCount());
  
              List<CompletableFuture<SqlResult>> futures = new ArrayList<>();
              for (i = 0; i < NUMBER_OF_QUERIES; ++i) {
                  if (i % 5 == 0) {
                      futures.add(this.session.sql("insert into testExecAsync (c1,c2,c9) values  (?,?,?)").bind(10).bind(l1 - 10).bind("a,d,c").executeAsync());
                  } else if (i % 5 == 1) {
                      futures.add(this.session.sql("REPLACE  DELAYED  into testExecAsync (c1,c2,c9) values  (?,?,?)").bind(10).bind(l1 - 100).bind("a,c")
                              .executeAsync());
                  } else if (i % 5 == 2) {
                      futures.add(this.session.sql("update  testExecAsync set c9 =? where c1 = (?+?+?)").bind("a,d,c,b").bind(3).bind(5).bind(2).executeAsync());
                  } else if (i % 5 == 3) {
                      futures.add(this.session.sql("select * from testExecAsync where c9&8 and c9&1 and c1 = (?+?)").bind(9).bind(1).executeAsync());
                  } else {
                      futures.add(this.session.sql("delete from testExecAsync where  c9 & ? and c9&1 and c1 = (?+?)").bind(8).bind(9).bind(1).executeAsync());
                  }
  
              }
  
              for (i = 0; i < NUMBER_OF_QUERIES; ++i) {
                  if (i % 5 == 0) {
                      sqlRes = futures.get(i).get();
                      assertEquals(1, sqlRes.getAffectedItemsCount());
                  } else if (i % 5 == 1) {
                      sqlRes = futures.get(i).get();
                      assertEquals(2, sqlRes.getAffectedItemsCount());
                  } else if (i % 5 == 2) {
                      sqlRes = futures.get(i).get();
                      assertEquals(1, sqlRes.getAffectedItemsCount());
                  } else if (i % 5 == 3) {
                      sqlRes = futures.get(i).get();
                      r = sqlRes.next();
                      assertEquals(10, r.getInt(0));
                      assertFalse(sqlRes.hasNext());
                  } else {
                      sqlRes = futures.get(i).get();
                      assertEquals(1, sqlRes.getAffectedItemsCount());
                  }
              }
          } finally {
              sqlUpdate("drop Procedure if exists testExecAsyncProc");
              sqlUpdate("drop table if exists testExecAsync");
          }
      }
  
      @Test
      public void testFetchOneFetchAllAsync() throws Exception {
          Row r = null;
          List<Row> rowList = null;
          try {
              CompletableFuture<SqlResult> asyncSqlRes = this.session.sql("drop table if exists testFetchOneFetchAllAsync").executeAsync();
              SqlResult sqlRes = asyncSqlRes.get();
              asyncSqlRes = this.session.sql("create table testFetchOneFetchAllAsync(a int,b bigint,c double,d blob)").executeAsync();
              sqlRes = asyncSqlRes.get();
              asyncSqlRes = this.session.sql("insert into testFetchOneFetchAllAsync values(?,?,?,?)").bind(1, 11).bind(21, "A").executeAsync();
              sqlRes = asyncSqlRes.get();
              asyncSqlRes = this.session.sql("insert into testFetchOneFetchAllAsync values(?,?,?,?)").bind(2, 12).bind(22, "B").executeAsync();
              sqlRes = asyncSqlRes.get();
              asyncSqlRes = this.session.sql("insert into testFetchOneFetchAllAsync values(?,?,?,?)").bind(3, 13).bind(23, "C").executeAsync();
              sqlRes = asyncSqlRes.get();
              asyncSqlRes = this.session.sql("insert into testFetchOneFetchAllAsync values(?,?,?,?)").bind(4, 14).bind(23, "D").executeAsync();
              sqlRes = asyncSqlRes.get();
  
              //With FetchOne()
              asyncSqlRes = this.session.sql("select * from testFetchOneFetchAllAsync where a<=? order by a asc").bind(5).executeAsync();
              SqlResult sqlRes1 = asyncSqlRes.get();
              int i = 0;
              while (sqlRes1.hasNext()) {
                  r = sqlRes1.fetchOne();
                  assertEquals((long) (i + 1), r.getInt(0));
                  i++;
              }
              assertThrows(WrongArgumentException.class, "Cannot fetchAll\\(\\) after starting iteration", () -> sqlRes1.fetchAll());
  
              asyncSqlRes = this.session.sql("select * from testFetchOneFetchAllAsync where a<=? order by a asc").bind(3).executeAsync();
              sqlRes = asyncSqlRes.get();
              rowList = sqlRes.fetchAll();
              assertEquals((long) 3, (long) rowList.size());
              for (i = 0; i < rowList.size(); i++) {
                  r = rowList.get(i);
                  assertEquals((long) (i + 1), r.getInt(0));
                  i++;
              }
          } finally {
              sqlUpdate("drop table if exists testFetchOneFetchAllAsync");
          }
      }
  
      /**
       * Few Negative Scenarios
       * 
       * @throws Exception
       */
      @Test
      public void testExecAsyncNegative() throws Exception {
          int i = 0;
          SqlResult sqlRes = null;
          Row r = null;
          try {
              assertThrows(ExecutionException.class, ".*Unknown table '" + this.schema.getName() + ".non_existing'.*", () -> {
                  CompletableFuture<SqlResult> res = this.session.sql("drop table non_existing").executeAsync();
                  res.get();
                  return null;
              });
  
              assertThrows(ExecutionException.class, ".* BIGINT value is out of range .*", () -> {
                  CompletableFuture<SqlResult> res = this.session.sql("select 123456*123456722323289").executeAsync();
                  res.get();
                  return null;
              });
  
              sqlUpdate("drop table if exists testExecAsyncNegative");
              sqlUpdate(
                      "create table testExecAsyncNegative(a int,b bigint ,c bigint  GENERATED ALWAYS AS (b*1000) VIRTUAL COMMENT '1',d bigint  GENERATED ALWAYS AS (c*100000) STOred  COMMENT '2')");
              sqlUpdate("Insert into  testExecAsyncNegative (a,b) values(1,100)");
              sqlUpdate("create index id on testExecAsyncNegative(d)");
              sqlUpdate("create unique index id2 on testExecAsyncNegative(a)");
  
              int NUMBER_OF_QUERIES = 5000;
              List<CompletableFuture<SqlResult>> futures = new ArrayList<>();
              for (i = 0; i < NUMBER_OF_QUERIES; ++i) {
                  if (i % 6 == 0) {
                      futures.add(this.session.sql("replace into testExecAsyncNegative (a,b) values(?,?)").bind(1).bind(1555666000000L).executeAsync());
                  } else if (i % 6 == 1) {
                      futures.add(this.session.sql("insert into testExecAsyncNegative (a,b) values (?,?) ON DUPLICATE KEY UPDATE b= ?").bind(1).bind(2)
                              .bind(1555666009990L).executeAsync());
                  } else if (i % 6 == 2) {
                      futures.add(this.session.sql("alter table testExecAsyncNegative add d point").executeAsync());
                  } else if (i % 6 == 3) {
                      futures.add(this.session.sql("insert into testExecAsyncNegative (a,b) values (?,?) ON DUPLICATE KEY UPDATE b=b/?").bind(1).bind(2).bind(0)
                              .executeAsync());
                  } else if (i % 6 == 4) {
                      futures.add(this.session.sql("SELECT /*+ max_execution_time (100) bad_hint */ SLEEP(0.5)").executeAsync());
                  } else {
                      futures.add(this.session.sql("select /*+*/ * from testExecAsyncNegative").executeAsync());
                  }
              }
  
              for (i = 0; i < NUMBER_OF_QUERIES; ++i) {
                  int i1 = i;
                  if (i % 6 == 0) {
                      assertThrows(ExecutionException.class, ".* BIGINT value is out of range .*", () -> futures.get(i1).get());
                  } else if (i % 6 == 1) {
                      assertThrows(ExecutionException.class, ".* BIGINT value is out of range .*", () -> futures.get(i1).get());
                  } else if (i % 6 == 2) {
                      assertThrows(ExecutionException.class, ".*Duplicate column name 'd'.*", () -> futures.get(i1).get());
                  } else if (i % 6 == 3) {
                      assertThrows(ExecutionException.class, ".*Division by 0.*", () -> futures.get(i1).get());
                  } else {
                      sqlRes = futures.get(i).get();
                      r = sqlRes.next();
                      assertEquals(1, r.getInt(0));
                      assertFalse(sqlRes.hasNext());
                      Iterator<Warning> w = sqlRes.getWarnings();
                      while (w.hasNext()) {
                          Warning element = w.next();
                          assertTrue(element.getMessage().contains("Optimizer hint syntax error"));
                      }
                  }
              }
          } finally {
              sqlUpdate("drop table if exists testExecAsyncNegative");
          }
      }
  
      /**
       * Test fix for Bug#97269 (30438500), POSSIBLE BUG IN COM.MYSQL.CJ.XDEVAPI.STREAMINGDOCRESULTBUILDER.
       * 
       * @throws Exception
       */
      @Test
      public void testBug97269() throws Exception {
          Session sess = null;
          try {
              String message1 = "W1";
              String message2 = "W2";
  
              // create notice message buffers
              Frame.Builder notice1 = Frame.newBuilder().setScope(Frame.Scope.LOCAL).setType(Frame.Type.WARNING_VALUE)
                      .setPayload(com.mysql.cj.x.protobuf.MysqlxNotice.Warning.newBuilder().setCode(MysqlErrorNumbers.ER_BAD_DB_ERROR).setMsg(message1).build()
                              .toByteString());
              Frame.Builder notice2 = Frame.newBuilder().setScope(Frame.Scope.GLOBAL).setType(Frame.Type.WARNING_VALUE)
                      .setPayload(com.mysql.cj.x.protobuf.MysqlxNotice.Warning.newBuilder().setCode(MysqlErrorNumbers.ER_BAD_DB_ERROR).setMsg(message2).build()
                              .toByteString());
  
              byte[] notice1Bytes = makeNoticeBytes(notice1.build());
              byte[] notice2Bytes = makeNoticeBytes(notice2.build());
              int size = notice1Bytes.length + notice2Bytes.length;
              byte[] noticesBytes = new byte[size];
              System.arraycopy(notice1Bytes, 0, noticesBytes, 0, notice1Bytes.length);
              System.arraycopy(notice2Bytes, 0, noticesBytes, notice1Bytes.length, notice2Bytes.length);
  
              InjectedSocketFactory.flushAllStaticData();
  
              String url = this.baseUrl + (this.baseUrl.contains("?") ? "" : "?")
                      + makeParam(PropertyKey.socketFactory, InjectedSocketFactory.class.getName(), !this.baseUrl.contains("?") || this.baseUrl.endsWith("?"))
                      + makeParam(PropertyKey.xdevapiSslMode, XdevapiSslMode.DISABLED.toString())
                      + makeParam(PropertyKey.xdevapiCompression, Compression.DISABLED.toString())
                      // to allow injection between result rows
                      + makeParam(PropertyKey.useReadAheadInput, "false");
  
              sess = this.fact.getSession(url);
              SocketFactory sf = ((SessionImpl) sess).getSession().getProtocol().getSocketConnection().getSocketFactory();
              assertTrue(InjectedSocketFactory.class.isAssignableFrom(sf.getClass()));
  
              Collection collection = sess.getDefaultSchema().createCollection("testBug97269");
              collection.add("{\"_id\":\"the_id\",\"g\":1}").execute();
  
              // StreamingDocResultBuilder
              InjectedSocketFactory.injectedBuffer = noticesBytes;
              DocResult docs = collection.find().fields("$._id as _id, $.g as g, 1 + 1 as q").execute();
              DbDoc doc = docs.next();
              assertEquals("the_id", ((JsonString) doc.get("_id")).getString());
              assertEquals(new Integer(1), ((JsonNumber) doc.get("g")).getInteger());
              assertEquals(new Integer(2), ((JsonNumber) doc.get("q")).getInteger());
  
              int cnt = 0;
              for (Iterator<Warning> warn = docs.getWarnings(); warn.hasNext();) {
                  Warning w = warn.next();
                  if (w.getMessage().equals(message1) || w.getMessage().equals(message2)) {
                      cnt++;
                  }
              }
              assertEquals(2, cnt);
  
              InjectedSocketFactory.flushAllStaticData();
              InjectedSocketFactory.injectedBuffer = noticesBytes;
  
              SqlResult rs1 = sess.sql("select 1").execute();
              assertEquals(1, rs1.fetchOne().getInt(0));
              cnt = 0;
              for (Iterator<Warning> warn = rs1.getWarnings(); warn.hasNext();) {
                  Warning w = warn.next();
                  if (w.getMessage().equals(message1) || w.getMessage().equals(message2)) {
                      cnt++;
                  }
              }
              assertEquals(2, cnt);
  
          } finally {
              InjectedSocketFactory.flushAllStaticData();
              dropCollection("testBug97269");
              if (sess != null) {
                  sess.close();
              }
          }
  
      }
  }
  