docs: @Override
    public Session create(CassandraStorage cassandra) {
      Closer closer = Closer.create();
      try {
        Cluster cluster = closer.register(buildCluster(cassandra));
        cluster.register(new QueryLogger.Builder().build());
        Session session;
        String keyspace = cassandra.keyspace();
        if (cassandra.ensureSchema()) {
          session = closer.register(cluster.connect());
          Schema.ensureExists(keyspace, cassandra.searchEnabled(), session);
          session.execute("USE " + keyspace);
        } else {
          LOG.debug("Skipping schema check on keyspace {} as ensureSchema was false", keyspace);
          session = cluster.connect(keyspace);
        }
  
        initializeUDTs(session);
  
        return session;
      } catch (RuntimeException e) {
        try {
          closer.close();
        } catch (IOException ignored) {
        }
        throw e;
      }
    }