public static Connection createConnection(URI connectionUri, Configuration conf,
    ExecutorService pool, final User user, Map<String, byte[]> connectionAttributes)
    throws IOException {
    Class<?> clazz = conf.getClass(ConnectionUtils.HBASE_CLIENT_CONNECTION_IMPL,
      ConnectionOverAsyncConnection.class, Connection.class);
    if (clazz != ConnectionOverAsyncConnection.class) {
      return TraceUtil.trace(() -> {
        try {
          // Default HCM#HCI is not accessible; make it so before invoking.
          Constructor<?> constructor = clazz.getDeclaredConstructor(Configuration.class,
            ExecutorService.class, User.class, ConnectionRegistry.class, Map.class);
          constructor.setAccessible(true);
          ConnectionRegistry registry = connectionUri != null
            ? ConnectionRegistryFactory.create(connectionUri, conf, user)
            : ConnectionRegistryFactory.create(conf, user);
          return user.runAs((PrivilegedExceptionAction<Connection>) () -> (Connection) constructor
            .newInstance(conf, pool, user, registry, connectionAttributes));
        } catch (NoSuchMethodException e) {
          LOG.debug("Constructor with connection registry not found for class {},"
            + " fallback to use old constructor", clazz.getName(), e);
        } catch (Exception e) {
          Throwables.throwIfInstanceOf(e, IOException.class);
          Throwables.throwIfUnchecked(e);
          throw new IOException(e);
        }

        try {
          // Default HCM#HCI is not accessible; make it so before invoking.
          Constructor<?> constructor = clazz.getDeclaredConstructor(Configuration.class,
            ExecutorService.class, User.class, Map.class);
          constructor.setAccessible(true);
          return user.runAs((PrivilegedExceptionAction<Connection>) () -> (Connection) constructor
            .newInstance(conf, pool, user, connectionAttributes));
        } catch (Exception e) {
          Throwables.throwIfInstanceOf(e, IOException.class);
          Throwables.throwIfUnchecked(e);
          throw new IOException(e);
        }
      }, () -> TraceUtil.createSpan(ConnectionFactory.class.getSimpleName() + ".createConnection"));
    } else {
      return FutureUtils.get(createAsyncConnection(connectionUri, conf, user, connectionAttributes))
        .toConnection();
    }
  }
Create a new Connection instance using the passed <code>conf</code> instance. Connection
encapsulates all housekeeping for a connection to the cluster. All tables and interfaces
created from returned connection share zookeeper connection(if used), meta cache, and
connections to region servers and masters. <br>
The caller is responsible for calling {@link Connection#close()} on the returned connection
instance. Typical usage:

<pre>
Connection connection = ConnectionFactory.createConnection(conf);
Table table = connection.getTable(TableName.valueOf("table1"));
try (Connection connection = ConnectionFactory.createConnection(conf);
  Table table = connection.getTable(TableName.valueOf("table1"))) {
  table.get(...);
  ...
}
</pre>

@param connectionUri        the connection uri for the hbase cluster
@param conf                 configuration
@param user                 the user the connection is for
@param pool                 the thread pool to use for batch operations
@param connectionAttributes attributes to be sent along to server during connection establish
@return Connection object for <code>conf</code>
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionFactory.java