/**
 * Gets the AccumuloConnector singleton, starting the MiniAccumuloCluster on initialization.
 * This singleton instance is required so all test cases access the same MiniAccumuloCluster.
 *
 * @return Accumulo connector
 */
public static Connector getAccumuloConnector()
{
    if (connector != null) {
        return connector;
    }

    try {
        MiniAccumuloCluster accumulo = createMiniAccumuloCluster();
        Instance instance = new ZooKeeperInstance(accumulo.getInstanceName(), accumulo.getZooKeepers());
        connector = instance.getConnector(MAC_USER, new PasswordToken(MAC_PASSWORD));
        LOG.info("Connection to MAC instance %s at %s established, user %s password %s", accumulo.getInstanceName(), accumulo.getZooKeepers(), MAC_USER, MAC_PASSWORD);
        return connector;
    }
    catch (AccumuloException | AccumuloSecurityException | InterruptedException | IOException e) {
        throw new PrestoException(UNEXPECTED_ACCUMULO_ERROR, "Failed to get connector to Accumulo", e);
    }
}
 

/**
 * Initializes the instance with a provided update interval.
 * 
 * @param connector
 *            A Connector to Accumulo
 * @param dateIndexTableName
 *            The name of the date index table
 * @param auths
 *            Any {@link Authorizations} to use
 */
protected DateIndexHelper initialize(Connector connector, Instance instance, String dateIndexTableName, Set<Authorizations> auths, int numQueryThreads,
                float collapseDatePercentThreshold) {
    this.connector = connector;
    this.instance = instance;
    this.dateIndexTableName = dateIndexTableName;
    this.auths = auths;
    this.numQueryThreads = numQueryThreads;
    this.collapseDatePercentThreshold = collapseDatePercentThreshold;
    
    if (log.isTraceEnabled()) {
        log.trace("Constructor  connector: " + (connector != null ? connector.getClass().getCanonicalName() : connector) + " with auths: " + auths
                        + " and date index table name: " + dateIndexTableName + "; " + numQueryThreads + " threads and " + collapseDatePercentThreshold
                        + " collapse date percent threshold");
    }
    return this;
}
 

public DataStore(Configuration conf) throws QonduitException {

        try {
            final HashMap<String, String> apacheConf = new HashMap<>();
            Configuration.Accumulo accumuloConf = conf.getAccumulo();
            apacheConf.put("instance.name", accumuloConf.getInstanceName());
            apacheConf.put("instance.zookeeper.host", accumuloConf.getZookeepers());
            final ClientConfiguration aconf = ClientConfiguration.fromMap(apacheConf);
            final Instance instance = new ZooKeeperInstance(aconf);
            connector = instance.getConnector(accumuloConf.getUsername(),
                    new PasswordToken(accumuloConf.getPassword()));
        } catch (Exception e) {
            throw new QonduitException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error creating DataStoreImpl",
                    e.getMessage(), e);
        }
    }
 

public static void main(String[] args) throws Exception {

        try (ConfigurableApplicationContext ctx = new SpringApplicationBuilder(SpringBootstrap.class)
                .bannerMode(Mode.OFF).web(WebApplicationType.NONE).run(args)) {
            Configuration conf = ctx.getBean(Configuration.class);

            final Map<String, String> properties = new HashMap<>();
            Accumulo accumuloConf = conf.getAccumulo();
            properties.put("instance.name", accumuloConf.getInstanceName());
            properties.put("instance.zookeeper.host", accumuloConf.getZookeepers());
            final ClientConfiguration aconf = ClientConfiguration.fromMap(properties);
            final Instance instance = new ZooKeeperInstance(aconf);
            Connector con = instance.getConnector(accumuloConf.getUsername(),
                    new PasswordToken(accumuloConf.getPassword()));
            Scanner s = con.createScanner(conf.getMetaTable(),
                    con.securityOperations().getUserAuthorizations(con.whoami()));
            try {
                s.setRange(new Range(Meta.METRIC_PREFIX, true, Meta.TAG_PREFIX, false));
                for (Entry<Key, Value> e : s) {
                    System.out.println(e.getKey().getRow().toString().substring(Meta.METRIC_PREFIX.length()));
                }
            } finally {
                s.close();
            }
        }
    }
 

public static void main(String[] args) throws Exception {
    try (ConfigurableApplicationContext ctx = new SpringApplicationBuilder(SpringBootstrap.class)
            .bannerMode(Banner.Mode.OFF).web(WebApplicationType.NONE).run(args)) {
        Configuration conf = ctx.getBean(Configuration.class);
        HashMap<String, String> apacheConf = new HashMap<>();
        Accumulo accumuloConf = conf.getAccumulo();
        apacheConf.put("instance.name", accumuloConf.getInstanceName());
        apacheConf.put("instance.zookeeper.host", accumuloConf.getZookeepers());
        ClientConfiguration aconf = ClientConfiguration.fromMap(apacheConf);
        Instance instance = new ZooKeeperInstance(aconf);
        Connector con = instance.getConnector(accumuloConf.getUsername(),
                new PasswordToken(accumuloConf.getPassword()));

        TabletMetadataQuery query = new TabletMetadataQuery(con, conf.getMetricsTable());
        TabletMetadataView view = query.run();

        System.out.println(view.toText(TimeUnit.DAYS));
    }
}
 

public MockRdfCloudStore() {
    super();
    final Instance instance = new MockInstance();
    try {
        final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
        setConf(conf);

        final Connector connector = instance.getConnector("", "");
        final AccumuloRyaDAO cdao = new AccumuloRyaDAO();
        cdao.setConf(conf);
        cdao.setConnector(connector);
        setRyaDAO(cdao);
        inferenceEngine = new InferenceEngine();
        inferenceEngine.setRyaDAO(cdao);
        inferenceEngine.setRefreshGraphSchedule(5000); //every 5 sec
        inferenceEngine.setConf(conf);
        setInferenceEngine(inferenceEngine);
    } catch (final Exception e) {
        e.printStackTrace();
    }
}
 

public MockRdfCloudStore() {
    super();
    Instance instance = new MockInstance();
    try {
        AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
        conf.setInfer(true);
        setConf(conf);
        Connector connector = instance.getConnector("", "");
        AccumuloRyaDAO cdao = new AccumuloRyaDAO();
        cdao.setConf(conf);
        cdao.setConnector(connector);
        setRyaDAO(cdao);
        inferenceEngine = new InferenceEngine();
        inferenceEngine.setRyaDAO(cdao);
        inferenceEngine.setRefreshGraphSchedule(5000); //every 5 sec
        inferenceEngine.setConf(conf);
        setInferenceEngine(inferenceEngine);
        internalInferenceEngine = inferenceEngine;
    } catch (Exception e) {
        e.printStackTrace();
    }
}
 

public MockRdfCloudStore() {
    super();
    Instance instance = new MockInstance();
    try {
        Connector connector = instance.getConnector("", "");
        setConf(new AccumuloRdfConfiguration());
        AccumuloRyaDAO cdao = new AccumuloRyaDAO();
        cdao.setConnector(connector);
        setRyaDAO(cdao);
        inferenceEngine = new InferenceEngine();
        inferenceEngine.setRyaDAO(cdao);
        inferenceEngine.setRefreshGraphSchedule(1000); //every sec
        setInferenceEngine(inferenceEngine);
    } catch (Exception e) {
        e.printStackTrace();
    }
}
 

protected String loadDataAndCreateQuery(final String sparql, final Collection<Statement> statements) throws Exception {
    requireNonNull(sparql);
    requireNonNull(statements);

    // Register the PCJ with Rya.
    final Instance accInstance = super.getAccumuloConnector().getInstance();
    final Connector accumuloConn = super.getAccumuloConnector();

    final RyaClient ryaClient = AccumuloRyaClientFactory.build(new AccumuloConnectionDetails(ACCUMULO_USER,
            ACCUMULO_PASSWORD.toCharArray(), accInstance.getInstanceName(), accInstance.getZooKeepers()), accumuloConn);

    final String pcjId = ryaClient.getCreatePCJ().createPCJ(RYA_INSTANCE_NAME, sparql, Sets.newHashSet(ExportStrategy.KAFKA));

    loadData(statements);

    // The PCJ Id is the topic name the results will be written to.
    return pcjId;
}
 

public Map<String, Serializable> getParams(final Configuration conf) {
    // get the configuration parameters
    final Instance instance = ConfigUtils.getInstance(conf);
    final String instanceId = instance.getInstanceName();
    final String zookeepers = instance.getZooKeepers();
    final String user = ConfigUtils.getUsername(conf);
    final String password = ConfigUtils.getPassword(conf);
    final String auths = ConfigUtils.getAuthorizations(conf).toString();
    final String tableName = getTableName(conf);
    final String tablePrefix = ConfigUtils.getTablePrefix(conf);

    final Map<String, Serializable> params = new HashMap<>();
    params.put("zookeeper", zookeepers);
    params.put("instance", instanceId);
    params.put("user", user);
    params.put("password", password);
    params.put("namespace", tableName);
    params.put("gwNamespace", tablePrefix + getClass().getSimpleName());

    params.put("Lock Management", LockManagementType.MEMORY.toString());
    params.put("Authorization Management Provider", AuthorizationManagementProviderType.EMPTY.toString());
    params.put("Authorization Data URL", null);
    params.put("Transaction Buffer Size", 10000);
    params.put("Query Index Strategy", QueryIndexStrategyType.HEURISTIC_MATCH.toString());
    return params;
}
 

/**
 * Creates the {@link DataStore} for the {@link GeoWaveGeoIndexer}.
 * @param conf the {@link Configuration}.
 * @return the {@link DataStore}.
 */
public DataStore createDataStore(final Configuration conf) throws IOException, GeoWavePluginException {
    final Map<String, Serializable> params = getParams(conf);
    final Instance instance = ConfigUtils.getInstance(conf);
    final boolean useMock = instance instanceof MockInstance;

    final StoreFactoryFamilySpi storeFactoryFamily;
    if (useMock) {
        storeFactoryFamily = new MemoryStoreFactoryFamily();
    } else {
        storeFactoryFamily = new AccumuloStoreFactoryFamily();
    }

    final GeoWaveGTDataStoreFactory geoWaveGTDataStoreFactory = new GeoWaveGTDataStoreFactory(storeFactoryFamily);
    final DataStore dataStore = geoWaveGTDataStoreFactory.createNewDataStore(params);

    return dataStore;
}
 

/**
 * Create a {@link Connector} that uses the provided connection details.
 *
 * @param username - The username the connection will use. (not null)
 * @param password - The password the connection will use. (not null)
 * @param instanceName - The name of the Accumulo instance. (not null)
 * @param zookeeperHostnames - A comma delimited list of the Zookeeper server hostnames. (not null)
 * @return A {@link Connector} that may be used to access the instance of Accumulo.
 * @throws AccumuloSecurityException Could not connect for security reasons.
 * @throws AccumuloException Could not connect for other reasons.
 */
public Connector connect(
        final String username,
        final CharSequence password,
        final String instanceName,
        final String zookeeperHostnames) throws AccumuloException, AccumuloSecurityException {
    requireNonNull(username);
    requireNonNull(password);
    requireNonNull(instanceName);
    requireNonNull(zookeeperHostnames);

    // Setup the password token that will be used.
    final PasswordToken token = new PasswordToken( password );

    // Connect to the instance of Accumulo.
    final Instance instance = new ZooKeeperInstance(instanceName, zookeeperHostnames);
    return instance.getConnector(username, token);
}
 

public static synchronized Connector getConnector(String instance, String zookeepers,
    String user, String pass) throws DataProviderException
{

  if (conn != null)
  {
    return conn;
  }
  if (checkMock())
  {
    return getMockConnector(instance, user, pass);
  }

  Instance inst = new ZooKeeperInstance(instance, zookeepers);
  try
  {
    conn = inst.getConnector(user, new PasswordToken(pass.getBytes()));
    return conn;
  }
  catch (Exception e)
  {
    throw new DataProviderException("problem creating connector", e);
  }
}
 

/**
 * For testing.
 *
 * @param instance
 * @param user
 * @param pass
 * @return
 */
public static Connector getMockConnector(String instance, String user,
    String pass) throws DataProviderException
{
  Instance mock = new MockInstance(instance);
  Connector conn = null;
  try
  {
    conn = mock.getConnector(user, pass.getBytes());
  }
  catch (Exception e)
  {
    throw new DataProviderException(
        "problem creating mock connector - " + e.getMessage(), e);
  }

  return conn;
}
 

public synchronized Connector getConnector(
    final String zookeeperUrl,
    final String instanceName,
    final String userName,
    final String password) throws AccumuloException, AccumuloSecurityException {

  final ConnectorConfig config =
      new ConnectorConfig(zookeeperUrl, instanceName, userName, password);
  Connector connector = connectorCache.get(config);
  if (connector == null) {
    final Instance inst = new ZooKeeperInstance(instanceName, zookeeperUrl);
    connector = inst.getConnector(userName, password);
    connectorCache.put(config, connector);
  }
  return connector;
}
 

public static void setQueryInfo(Job job, Set<String> entityTypes, Node query, EntityShardBuilder shardBuilder, TypeRegistry<String> typeRegistry) throws AccumuloSecurityException, AccumuloException, TableNotFoundException, IOException {

        validateOptions(job);

        checkNotNull(shardBuilder);
        checkNotNull(query);
        checkNotNull(entityTypes);
        checkNotNull(typeRegistry);

        Instance instance = getInstance(job);
        Connector connector = instance.getConnector(getPrincipal(job), getAuthenticationToken(job));
        BatchScanner scanner = connector.createBatchScanner(DEFAULT_IDX_TABLE_NAME, getScanAuthorizations(job), 5);
        GlobalIndexVisitor globalIndexVisitor = new EntityGlobalIndexVisitor(scanner, shardBuilder, entityTypes);

        configureScanner(job, entityTypes, query, new NodeToJexl(typeRegistry), globalIndexVisitor, typeRegistry, OptimizedQueryIterator.class);

        job.getConfiguration().setBoolean(QUERY, true);
        job.getConfiguration().set(TYPE_REGISTRY, new String(toBase64(typeRegistry)));
    }
 

public static void main(String args[]) throws AccumuloSecurityException, AccumuloException, TableNotFoundException {

        if (args.length != 7) {
            System.out.println("Usage: " + DailyShardSplitter.class.getName() + "<zookeepers> <instance> <username> <password> <tableName> <start day: yyyy-mm-dd> <stop day: yyyy-mm-dd>");
            System.exit(1);
        }

        String zookeepers = args[0];
        String instance = args[1];
        String username = args[2];
        String password = args[3];
        String tableName = args[4];
        DateTime start = DateTime.parse(args[5]);
        DateTime stop = DateTime.parse(args[6]);


        Instance accInst = new ZooKeeperInstance(instance, zookeepers);
        Connector connector = accInst.getConnector(username, password.getBytes());

        SortedSet<Text> shards = new DailyShardBuilder(Constants.DEFAULT_PARTITION_SIZE)
                .buildShardsInRange(start.toDate(), stop.toDate());

        connector.tableOperations().addSplits(tableName, shards);
    }
 

@Override
public Connector get()
{
    try {
        Instance inst = new ZooKeeperInstance(instance, zooKeepers);
        Connector connector = inst.getConnector(username, new PasswordToken(password.getBytes(UTF_8)));
        LOG.info("Connection to instance %s at %s established, user %s", instance, zooKeepers, username);
        return connector;
    }
    catch (AccumuloException | AccumuloSecurityException e) {
        throw new PrestoException(UNEXPECTED_ACCUMULO_ERROR, "Failed to get connector to Accumulo", e);
    }
}
 

public AccumuloIndexAgeDisplay(Instance instance, String tableName, String columns, String userName, PasswordToken password, Integer[] buckets)
                throws AccumuloException, AccumuloSecurityException {
    this.tableName = tableName;
    setColumns(columns);
    this.userName = userName;
    setBuckets(buckets);
    
    conn = instance.getConnector(userName, password);
}
 

public AccumuloConfiguration(Instance instance, String accumuloUser, String accumuloPassword,
                             boolean isMock) throws AccumuloSecurityException, IOException {
  //NOTE: new Job(new Configuration()) does not work in scala shell due to the toString method's implementation
  //to get it to work in scala override the toString method and it will work
  
  //initialize fields, these are needed for lazy initialization of connector
  this.zkInstance = instance;
  this.accumuloUser = accumuloUser;
  this.accumuloPassword = accumuloPassword;
  
  this.job = new Job(new Configuration());
  AbstractInputFormat.setConnectorInfo(job, accumuloUser, new PasswordToken(accumuloPassword));
  AccumuloOutputFormat.setConnectorInfo(job, accumuloUser, new PasswordToken(accumuloPassword));

  AbstractInputFormat.setScanAuthorizations(job, new Authorizations());

  if (isMock) {
    AbstractInputFormat.setMockInstance(job, instance.getInstanceName());
    AccumuloOutputFormat.setMockInstance(job, instance.getInstanceName());
  } else {

    this.clientConfig = new ClientConfiguration();
    this.clientConfig.withInstance(instance.getInstanceName());
    this.clientConfig.withZkHosts(instance.getZooKeepers());

    AbstractInputFormat.setZooKeeperInstance(job, clientConfig);
    AccumuloOutputFormat.setZooKeeperInstance(job, this.clientConfig);
  }
}
 

@Before
public void init() throws Exception {
    conf = getConf();
    Instance mock = new MockInstance("instance");
    Connector conn = mock.getConnector("root", new PasswordToken(""));
    dao = new AccumuloRyaDAO();
    dao.setConnector(conn);
    dao.init();
    eval = new ParallelEvaluationStrategyImpl(new StoreTripleSource(conf, dao), null, null, conf);
}
 

@Override
public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
    //read the params from AccumuloInputFormat
    Configuration conf = jobContext.getConfiguration();
    Instance instance = MRUtils.AccumuloProps.getInstance(jobContext);
    String user = MRUtils.AccumuloProps.getUsername(jobContext);
    AuthenticationToken password = MRUtils.AccumuloProps.getPassword(jobContext);
    String table = MRUtils.AccumuloProps.getTablename(jobContext);
    ArgumentChecker.notNull(instance);
    ArgumentChecker.notNull(table);

    //find the files necessary
    try {
        Connector connector = instance.getConnector(user, password);
        TableOperations tos = connector.tableOperations();
        String tableId = tos.tableIdMap().get(table);
        Scanner scanner = connector.createScanner("accumulo.metadata", Authorizations.EMPTY); //TODO: auths?
        scanner.setRange(new Range(new Text(tableId + "\u0000"), new Text(tableId + "\uFFFD")));
        scanner.fetchColumnFamily(new Text("file"));
        List<String> files = new ArrayList<String>();
        List<InputSplit> fileSplits = new ArrayList<InputSplit>();
        for (Map.Entry<Key, Value> entry : scanner) {
            String file = entry.getKey().getColumnQualifier().toString();
            Path path = new Path(file);
            FileSystem fs = path.getFileSystem(conf);
            FileStatus fileStatus = fs.getFileStatus(path);
            long len = fileStatus.getLen();
            BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatus, 0, len);
            files.add(file);
            fileSplits.add(new FileSplit(path, 0, len, fileBlockLocations[0].getHosts()));
        }
        System.out.println(files);
        return fileSplits;
    } catch (Exception e) {
        throw new IOException(e);
    }
}
 

private static Connector createAccumuloConnector(final PcjAdminClientProperties clientProps) throws AccumuloException, AccumuloSecurityException {
    checkNotNull(clientProps);

    // Connect to the Zookeepers.
    final String instanceName = clientProps.getAccumuloInstance();
    final String zooServers = clientProps.getAccumuloZookeepers();
    final Instance inst = new ZooKeeperInstance(instanceName, zooServers);

    // Create a connector to the Accumulo that hosts the PCJ export tables.
    return inst.getConnector(clientProps.getAccumuloUsername(), new PasswordToken(clientProps.getAccumuloPassword()));
}
 

@Override
public Optional<IncrementalResultExporter> build(Context context) throws IncrementalExporterFactoryException, ConfigurationException {
    checkNotNull(context);

    // Wrap the context's parameters for parsing.
    final RyaExportParameters params = new RyaExportParameters( context.getObserverConfiguration().toMap() );

    if(params.getUsePeriodicBindingSetExporter()) {
        // Setup Zookeeper connection info.
        final String accumuloInstance = params.getAccumuloInstanceName().get();
        final String zookeeperServers =  params.getZookeeperServers().get().replaceAll(";", ",");
        final Instance inst = new ZooKeeperInstance(accumuloInstance, zookeeperServers);

        try {
            // Setup Accumulo connection info.
            final String exporterUsername = params.getExporterUsername().get();
            final String exporterPassword = params.getExporterPassword().get();
            final Connector accumuloConn = inst.getConnector(exporterUsername, new PasswordToken(exporterPassword));

            // Setup Rya PCJ Storage.
            final String ryaInstanceName = params.getRyaInstanceName().get();
            final PeriodicQueryResultStorage periodicStorage = new AccumuloPeriodicQueryResultStorage(accumuloConn, ryaInstanceName);
            
            // Make the exporter.
            final IncrementalBindingSetExporter exporter = new PeriodicBindingSetExporter(periodicStorage);
            return Optional.of(exporter);

        } catch (final AccumuloException | AccumuloSecurityException e) {
            throw new IncrementalExporterFactoryException("Could not initialize the Accumulo connector using the provided configuration.", e);
        }
    } else {
        return Optional.absent();
    }
}
 

@Override
public Optional<IncrementalResultExporter> build(final Context context) throws IncrementalExporterFactoryException, ConfigurationException {
    checkNotNull(context);
    // Wrap the context's parameters for parsing.
    final RyaExportParameters params = new RyaExportParameters( context.getObserverConfiguration().toMap() );

    if(params.getUseRyaBindingSetExporter()) {
        // Setup Zookeeper connection info.
        final String accumuloInstance = params.getAccumuloInstanceName().get();
        final String zookeeperServers =  params.getZookeeperServers().get().replaceAll(";", ",");
        final Instance inst = new ZooKeeperInstance(accumuloInstance, zookeeperServers);

        try {
            // Setup Accumulo connection info.
            final String exporterUsername = params.getExporterUsername().get();
            final String exporterPassword = params.getExporterPassword().get();
            final Connector accumuloConn = inst.getConnector(exporterUsername, new PasswordToken(exporterPassword));

            // Setup Rya PCJ Storage.
            final String ryaInstanceName = params.getRyaInstanceName().get();
            final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(accumuloConn, ryaInstanceName);

            // Make the exporter.
            final IncrementalBindingSetExporter exporter = new RyaBindingSetExporter(pcjStorage);
            return Optional.of(exporter);

        } catch (final AccumuloException | AccumuloSecurityException e) {
            throw new IncrementalExporterFactoryException("Could not initialize the Accumulo connector using the provided configuration.", e);
        }
    } else {
        return Optional.absent();
    }
}
 

public void runTest(final String sparql, final Collection<Statement> statements, final Collection<BindingSet> expectedResults) throws Exception {
    requireNonNull(sparql);
    requireNonNull(statements);
    requireNonNull(expectedResults);

    // Register the PCJ with Rya.
    final Instance accInstance = super.getAccumuloConnector().getInstance();
    final Connector accumuloConn = super.getAccumuloConnector();

    final RyaClient ryaClient = AccumuloRyaClientFactory.build(new AccumuloConnectionDetails(
            getUsername(),
            getPassword().toCharArray(),
            accInstance.getInstanceName(),
            accInstance.getZooKeepers()), accumuloConn);

    ryaClient.getCreatePCJ().createPCJ(getRyaInstanceName(), sparql);

    // Write the data to Rya.
    final SailRepositoryConnection ryaConn = super.getRyaSailRepository().getConnection();
    ryaConn.begin();
    ryaConn.add(statements);
    ryaConn.commit();
    ryaConn.close();

    // Wait for the Fluo application to finish computing the end result.
    super.getMiniFluo().waitForObservers();

    // Fetch the value that is stored within the PCJ table.
    try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(accumuloConn, getRyaInstanceName())) {
        final String pcjId = pcjStorage.listPcjs().get(0);
        final Set<BindingSet> results = Sets.newHashSet( pcjStorage.listResults(pcjId) );

        // Ensure the result of the query matches the expected result.
        assertEquals(expectedResults, results);
    }
}
 

@BeforeClass
public static void beforeClass() throws Exception {
    Logger.getLogger(ClientCnxn.class).setLevel(Level.ERROR);

    // Setup and start the Mini Accumulo.
    cluster = clusterInstance.getCluster();

    // Store a connector to the Mini Accumulo.
    instanceName = cluster.getInstanceName();
    zookeepers = cluster.getZooKeepers();

    final Instance instance = new ZooKeeperInstance(instanceName, zookeepers);
    accumuloConn = instance.getConnector(clusterInstance.getUsername(), new PasswordToken(clusterInstance.getPassword()));
}
 

/**
 * Setup a Mini Accumulo cluster that uses a temporary directory to store its data.
 *
 * @return A Mini Accumulo cluster.
 */
private static MiniAccumuloCluster startMiniAccumulo() throws IOException, InterruptedException, AccumuloException, AccumuloSecurityException {
    final File miniDataDir = Files.createTempDir();

    // Setup and start the Mini Accumulo.
    final MiniAccumuloCluster accumulo = new MiniAccumuloCluster(miniDataDir, "password");
    accumulo.start();

    // Store a connector to the Mini Accumulo.
    final Instance instance = new ZooKeeperInstance(accumulo.getInstanceName(), accumulo.getZooKeepers());
    accumuloConn = instance.getConnector("root", new PasswordToken("password"));

    return accumulo;
}
 

/**
 * Create an {@link Instance} that may be used to create {@link Connector}s
 * to Accumulo. If the configuration has the {@link #USE_MOCK_INSTANCE} flag
 * set, then the instance will be be a {@link MockInstance} instead of a
 * Zookeeper backed instance.
 *
 * @param conf - The configuration object that will be interrogated. (not null)
 * @return The {@link Instance} that may be used to connect to Accumulo.
 */
public static Instance getInstance(final Configuration conf) {
    // Pull out the Accumulo specific configuration values.
    final AccumuloRdfConfiguration accConf = new AccumuloRdfConfiguration(conf);
    final String instanceName = accConf.getInstanceName();
    final String zoookeepers = accConf.getZookeepers();

    // Create an Instance a mock if the mock flag is set.
    if (useMockInstance(conf)) {
        return new MockInstance(instanceName);
    }

    // Otherwise create an Instance to a Zookeeper managed instance of Accumulo.
    return new ZooKeeperInstance(instanceName, zoookeepers);
}
 

@BeforeClass
public static void beforeClass() throws Exception {
    Logger.getLogger(ClientCnxn.class).setLevel(Level.ERROR);

    // Setup and start the Mini Accumulo.
    cluster = clusterInstance.getCluster();

    // Store a connector to the Mini Accumulo.
    instanceName = cluster.getInstanceName();
    zookeepers = cluster.getZooKeepers();

    final Instance instance = new ZooKeeperInstance(instanceName, zookeepers);
    accumuloConn = instance.getConnector(clusterInstance.getUsername(), new PasswordToken(clusterInstance.getPassword()));
}