presto-accumulo/src/main/java/io/prestosql/plugin/accumulo/metadata/ZooKeeperMetadataManager.java/ZooKeeperMetadataManager/ZooKeeperMetadataManager
class ZooKeeperMetadataManager:    @Inject
    public ZooKeeperMetadataManager(AccumuloConfig config, TypeManager typeManager)
    {
        requireNonNull(typeManager, "typeManager is null");

        // Create JSON deserializer for the AccumuloTable
        ObjectMapperProvider objectMapperProvider = new ObjectMapperProvider();
        objectMapperProvider.setJsonDeserializers(ImmutableMap.of(Type.class, new AccumuloModule.TypeDeserializer(typeManager)));
        mapper = objectMapperProvider.get();

        String zkMetadataRoot = config.getZkMetadataRoot();
        String zookeepers = config.getZooKeepers();

        // Create the connection to ZooKeeper to check if the metadata root exists
        CuratorFramework checkRoot = CuratorFrameworkFactory.newClient(zookeepers, new RetryForever(1000));
        checkRoot.start();

        try {
            // If the metadata root does not exist, create it
            if (checkRoot.checkExists().forPath(zkMetadataRoot) == null) {
                checkRoot.create().forPath(zkMetadataRoot);
            }
        }
        catch (Exception e) {
            throw new PrestoException(ZOOKEEPER_ERROR, "ZK error checking metadata root", e);
        }
        checkRoot.close();

        // Create the curator client framework to use for metadata management, set at the ZK root
        curator = CuratorFrameworkFactory.newClient(zookeepers + zkMetadataRoot, new RetryForever(1000));
        curator.start();

        try {
            // Create default schema should it not exist
            if (curator.checkExists().forPath("/" + DEFAULT_SCHEMA) == null) {
                curator.create().forPath("/" + DEFAULT_SCHEMA);
            }
        }
        catch (Exception e) {
            throw new PrestoException(ZOOKEEPER_ERROR, "ZK error checking/creating default schema", e);
        }
    }