import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
@Test
public void getVolumesTest() throws Exception {
	MiniAccumuloClusterImpl accumulo = this.getDoubleVolumeMAC();
	try {
		String expectedVols = accumulo.getConfig().getSiteConfig().get(Property.INSTANCE_VOLUMES.getKey());
		Assert.assertNotNull(expectedVols);
		Configuration cfg = this.getConfigurationForMac(accumulo);
		Volume[] vols = Utils.getVolumes(cfg);
		
		List<String> expected = Arrays.asList(expectedVols.split(","));
		Collections.sort(expected);
		
		List<String> actual = new ArrayList<String>();
		for (Volume vol : vols) {
			actual.add(vol.getFileSystem().getScheme() + ":" + vol.getBasePath());
		}
		Collections.sort(actual);
		Assert.assertEquals(expected,actual);
	} finally {
		accumulo.stop();
	}
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
@Override
public void initializeSecurity(TCredentials credentials, String principal,
    byte[] token) throws AccumuloSecurityException {
  String pass = null;
  SiteConfiguration siteconf = SiteConfiguration.getInstance(
      DefaultConfiguration.getInstance());
  String jksFile = siteconf.get(
      Property.GENERAL_SECURITY_CREDENTIAL_PROVIDER_PATHS);

  if (jksFile == null) {
    throw new RuntimeException(
        Property.GENERAL_SECURITY_CREDENTIAL_PROVIDER_PATHS +
            " not specified in accumulo-site.xml");
  }
  try {
    pass = new String(ProviderUtil.getPassword(jksFile,
        ROOT_INITIAL_PASSWORD_PROPERTY));
  } catch (IOException ioe) {
    throw new RuntimeException("Can't get key " +
        ROOT_INITIAL_PASSWORD_PROPERTY + " from " + jksFile, ioe);
  }
  zkAuthenticator.initializeSecurity(credentials, principal,
      pass.getBytes(Charset.forName("UTF-8")));
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
private void startMiniAccumulo(
		final MiniAccumuloConfigImpl config )
		throws IOException,
		InterruptedException {

	final LinkedList<String> jvmArgs = new LinkedList<>();
	jvmArgs.add("-XX:CompressedClassSpaceSize=512m");
	jvmArgs.add("-XX:MaxMetaspaceSize=512m");
	jvmArgs.add("-Xmx512m");

	Runtime.getRuntime().addShutdownHook(
			new Thread() {
				@Override
				public void run() {
					tearDown();
				}
			});
	final Map<String, String> siteConfig = config.getSiteConfig();
	siteConfig.put(
			Property.INSTANCE_ZK_HOST.getKey(),
			zookeeper);
	config.setSiteConfig(siteConfig);
	miniAccumulo.start();
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
@Test
public void testBloomFilters() throws Exception {
  String tableName = getUniqueNames(1)[0];
  c.tableOperations().create(tableName);
  c.tableOperations().setProperty(tableName, Property.TABLE_BLOOM_ENABLED.getKey(), "true");
  String[] args = new String[] {"--seed", "7", "-c", getConnectionFile(), "--num", "100000", "--min", "0", "--max", "1000000000", "--size", "50",
      "--batchMemory", "2M", "--batchLatency", "60", "--batchThreads", "3", "-t", tableName};

  goodExec(RandomBatchWriter.class, args);
  c.tableOperations().flush(tableName, null, null, true);
  long diff = 0, diff2 = 0;
  // try the speed test a couple times in case the system is loaded with other tests
  for (int i = 0; i < 2; i++) {
    long now = System.currentTimeMillis();
    args = new String[] {"--seed", "7", "-c", getConnectionFile(), "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50", "--scanThreads",
        "4", "-t", tableName};
    goodExec(RandomBatchScanner.class, args);
    diff = System.currentTimeMillis() - now;
    now = System.currentTimeMillis();
    args = new String[] {"--seed", "8", "-c", getConnectionFile(), "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50", "--scanThreads",
        "4", "-t", tableName};
    int retCode = getClusterControl().exec(RandomBatchScanner.class, args);
    assertEquals(1, retCode);
    diff2 = System.currentTimeMillis() - now;
    if (diff2 < diff)
      break;
  }
  assertTrue(diff2 < diff);
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
public MiniAccumuloClusterImpl getDoubleVolumeMAC() throws Exception {
	File mac = getTestDir("mac");
	String dfsURI = new File(mac, "hdfs-mac").toURI().toString();
	String dirs = dfsURI + "/accumulo-a," + dfsURI + "/accumulo-b";
	MiniAccumuloConfigImpl cfg = new MiniAccumuloConfigImpl(mac, "");
	cfg.useMiniDFS(true);
	cfg.setProperty(Property.INSTANCE_VOLUMES, dirs);
	MiniAccumuloClusterImpl accumulo = new MiniAccumuloClusterImpl(cfg);
	accumulo.start();
	return accumulo;
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
/**
 * Creates a table for Gaffer data and enables the correct Bloom filter;
 * removes the versioning iterator and adds an aggregator Iterator the
 * {@link org.apache.accumulo.core.iterators.user.AgeOffFilter} for the
 * specified time period.
 *
 * @param store the accumulo store
 * @throws StoreException       failure to create accumulo connection or add iterator settings
 * @throws TableExistsException failure to create table
 */
public static synchronized void createTable(final AccumuloStore store)
        throws StoreException, TableExistsException {
    // Create table
    final String tableName = store.getTableName();
    if (null == tableName) {
        throw new AccumuloRuntimeException("Table name is required.");
    }
    final Connector connector = store.getConnection();
    if (connector.tableOperations().exists(tableName)) {
        LOGGER.info("Table {} exists, not creating", tableName);
        return;
    }
    try {
        LOGGER.info("Creating table {} as user {}", tableName, connector.whoami());
        connector.tableOperations().create(tableName);
        final String repFactor = store.getProperties().getTableFileReplicationFactor();
        if (null != repFactor) {
            LOGGER.info("Table file replication set to {} on table {}", repFactor, tableName);
            connector.tableOperations().setProperty(tableName, Property.TABLE_FILE_REPLICATION.getKey(), repFactor);
        }

        // Enable Bloom filters using ElementFunctor
        LOGGER.info("Enabling Bloom filter on table {}", tableName);
        connector.tableOperations().setProperty(tableName, Property.TABLE_BLOOM_ENABLED.getKey(), "true");
        connector.tableOperations().setProperty(tableName, Property.TABLE_BLOOM_KEY_FUNCTOR.getKey(),
                store.getKeyPackage().getKeyFunctor().getClass().getName());

        // Remove versioning iterator from table for all scopes
        LOGGER.info("Removing versioning iterator from table {}", tableName);
        final EnumSet<IteratorScope> iteratorScopes = EnumSet.allOf(IteratorScope.class);
        connector.tableOperations().removeIterator(tableName, "vers", iteratorScopes);

        if (store.getSchema().isAggregationEnabled()) {
            // Add Combiner iterator to table for all scopes
            LOGGER.info("Adding Aggregator iterator to table {} for all scopes", tableName);
            connector.tableOperations().attachIterator(tableName,
                    store.getKeyPackage().getIteratorFactory().getAggregatorIteratorSetting(store));
        } else {
            LOGGER.info("Aggregator iterator has not been added to table {}", tableName);
        }

        if (store.getProperties().getEnableValidatorIterator()) {
            // Add validator iterator to table for all scopes
            final IteratorSetting itrSetting = store.getKeyPackage().getIteratorFactory().getValidatorIteratorSetting(store);
            if (null == itrSetting) {
                LOGGER.info("Not adding Validator iterator to table {} as there are no validation functions defined in the schema", tableName);
            } else {
                LOGGER.info("Adding Validator iterator to table {} for all scopes", tableName);
                connector.tableOperations().attachIterator(tableName,
                        store.getKeyPackage().getIteratorFactory().getValidatorIteratorSetting(store));
            }
        } else {
            LOGGER.info("Validator iterator has not been added to table {}", tableName);
        }
    } catch (final AccumuloSecurityException | TableNotFoundException | AccumuloException | IteratorSettingException e) {
        throw new StoreException(e.getMessage(), e);
    }
    setLocalityGroups(store);
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
@Test
public void shouldCreateTableCorrectlyIfSchemaContainsNoAggregators() throws Exception {
    // Given
    final SingleUseMockAccumuloStore store = new SingleUseMockAccumuloStore();
    final Schema schema = new Schema.Builder()
            .type(TestTypes.ID_STRING, new TypeDefinition.Builder()
                    .clazz(String.class)
                    .validateFunctions(new Exists())
                    .build())
            .type(TestTypes.DIRECTED_TRUE, Boolean.class)
            .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder()
                    .source(TestTypes.ID_STRING)
                    .destination(TestTypes.ID_STRING)
                    .directed(TestTypes.DIRECTED_TRUE)
                    .aggregate(false)
                    .build())
            .build();

    final AccumuloProperties props = AccumuloProperties.loadStoreProperties(StreamUtil.storeProps(TableUtilsTest.class));
    store.initialise(NO_AGGREGATORS_GRAPH_ID, schema, props);

    // When
    TableUtils.createTable(store);

    // Then
    final Map<String, EnumSet<IteratorScope>> itrs = store.getConnection().tableOperations().listIterators(NO_AGGREGATORS_GRAPH_ID);
    assertEquals(1, itrs.size());

    final EnumSet<IteratorScope> validator = itrs.get(AccumuloStoreConstants.VALIDATOR_ITERATOR_NAME);
    assertEquals(EnumSet.allOf(IteratorScope.class), validator);
    final IteratorSetting validatorSetting = store.getConnection().tableOperations().getIteratorSetting(NO_AGGREGATORS_GRAPH_ID, AccumuloStoreConstants.VALIDATOR_ITERATOR_NAME, IteratorScope.majc);
    assertEquals(AccumuloStoreConstants.VALIDATOR_ITERATOR_PRIORITY, validatorSetting.getPriority());
    assertEquals(ValidatorFilter.class.getName(), validatorSetting.getIteratorClass());
    final Map<String, String> validatorOptions = validatorSetting.getOptions();
    assertNotNull(Schema.fromJson(validatorOptions.get(AccumuloStoreConstants.SCHEMA).getBytes(CommonConstants.UTF_8)).getEdge(TestGroups.EDGE));
    assertEquals(ByteEntityAccumuloElementConverter.class.getName(), validatorOptions.get(AccumuloStoreConstants.ACCUMULO_ELEMENT_CONVERTER_CLASS));

    final EnumSet<IteratorScope> aggregator = itrs.get(AccumuloStoreConstants.AGGREGATOR_ITERATOR_NAME);
    assertNull(aggregator);
    final IteratorSetting aggregatorSetting = store.getConnection().tableOperations().getIteratorSetting(NO_AGGREGATORS_GRAPH_ID, AccumuloStoreConstants.AGGREGATOR_ITERATOR_NAME, IteratorScope.majc);
    assertNull(aggregatorSetting);

    final Map<String, String> tableProps = new HashMap<>();
    for (final Map.Entry<String, String> entry : store.getConnection()
            .tableOperations().getProperties(NO_AGGREGATORS_GRAPH_ID)) {
        tableProps.put(entry.getKey(), entry.getValue());
    }

    assertEquals(0, Integer.parseInt(tableProps.get(Property.TABLE_FILE_REPLICATION.getKey())));
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
public static void main(String[] args) throws Exception {

    Parameters params = new Parameters();
    JCommander jc = new JCommander(params);

    try {
      jc.parse(args);
      if (params.args == null || params.args.size() != 2)
        throw new ParameterException("Expected two arguments");
    } catch (ParameterException pe) {
      System.out.println(pe.getMessage());
      jc.setProgramName(Mini.class.getSimpleName());
      jc.usage();
      System.exit(-1);
    }

    MiniAccumuloConfig cfg = new MiniAccumuloConfig(new File(params.args.get(0)), "secret");
    cfg.setZooKeeperPort(params.zookeeperPort);
    cfg.setNumTservers(params.tabletServers);
    if (params.moreMemory) {
      cfg.setMemory(ServerType.TABLET_SERVER, 2, MemoryUnit.GIGABYTE);
      Map<String, String> site = new HashMap<>();
      site.put(Property.TSERV_DATACACHE_SIZE.getKey(), "768M");
      site.put(Property.TSERV_INDEXCACHE_SIZE.getKey(), "256M");
      cfg.setSiteConfig(site);
    }

    MiniAccumuloCluster cluster = new MiniAccumuloCluster(cfg);
    cluster.start();

    FluoConfiguration fluoConfig = new FluoConfiguration();

    fluoConfig.setMiniStartAccumulo(false);
    fluoConfig.setAccumuloInstance(cluster.getInstanceName());
    fluoConfig.setAccumuloUser("root");
    fluoConfig.setAccumuloPassword("secret");
    fluoConfig.setAccumuloZookeepers(cluster.getZooKeepers());
    fluoConfig.setInstanceZookeepers(cluster.getZooKeepers() + "/fluo");

    fluoConfig.setAccumuloTable("data");
    fluoConfig.setWorkerThreads(params.workerThreads);

    fluoConfig.setApplicationName("phrasecount");

    Application.configure(fluoConfig, new Application.Options(17, 17, cluster.getInstanceName(),
        cluster.getZooKeepers(), "root", "secret", "pcExport"));

    FluoFactory.newAdmin(fluoConfig).initialize(new InitializationOptions());

    MiniFluo miniFluo = FluoFactory.newMiniFluo(fluoConfig);

    miniFluo.getClientConfiguration().save(new File(params.args.get(1)));

    System.out.println();
    System.out.println("Wrote : " + params.args.get(1));
  }
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
public static void main(
		final String[] args )
		throws Exception {
	org.apache.log4j.Logger.getRootLogger().setLevel(
			org.apache.log4j.Level.WARN);

	final boolean interactive = (System.getProperty("interactive") != null) ? Boolean.parseBoolean(System
			.getProperty("interactive")) : true;

	final String password = System.getProperty(
			"password",
			"secret");

	final File tempDir = Files.createTempDir();
	final String instanceName = System.getProperty(
			"instanceName",
			"accumulo");
	final MiniAccumuloConfigImpl miniAccumuloConfig = new MiniAccumuloConfigImpl(
			tempDir,
			password).setNumTservers(
			2).setInstanceName(
			instanceName).setZooKeeperPort(
			2181);

	miniAccumuloConfig.setProperty(
			Property.MONITOR_PORT,
			"50095");

	final MiniAccumuloClusterImpl accumulo = MiniAccumuloClusterFactory.newAccumuloCluster(
			miniAccumuloConfig,
			AccumuloMiniCluster.class);
	accumulo.start();

	accumulo.exec(Monitor.class);

	System.out.println("starting up ...");
	Thread.sleep(3000);

	System.out.println("cluster running with instance name " + accumulo.getInstanceName() + " and zookeepers "
			+ accumulo.getZooKeepers());

	if (interactive) {
		System.out.println("hit Enter to shutdown ..");
		System.in.read();
		System.out.println("Shutting down!");
		accumulo.stop();
	}
	else {
		Runtime.getRuntime().addShutdownHook(
				new Thread() {
					@Override
					public void run() {
						try {
							accumulo.stop();
						}
						catch (final Exception e) {
							LOGGER.warn(
									"Unable to shutdown accumulo",
									e);
							System.out.println("Error shutting down accumulo.");
						}
						System.out.println("Shutting down!");
					}
				});

		while (true) {
			Thread.sleep(TimeUnit.MILLISECONDS.convert(
					Long.MAX_VALUE,
					TimeUnit.DAYS));
		}
	}
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
private void startMiniAccumulo(
		final MiniAccumuloConfigImpl config )
		throws IOException,
		InterruptedException {

	final LinkedList<String> jvmArgs = new LinkedList<>();
	jvmArgs.add("-XX:CompressedClassSpaceSize=512m");
	jvmArgs.add("-XX:MaxMetaspaceSize=512m");
	jvmArgs.add("-Xmx512m");

	Runtime.getRuntime().addShutdownHook(
			new Thread() {
				@Override
				public void run() {
					tearDown();
				}
			});
	final Map<String, String> siteConfig = config.getSiteConfig();
	siteConfig.put(
			Property.INSTANCE_ZK_HOST.getKey(),
			zookeeper);
	config.setSiteConfig(siteConfig);

	final LinkedList<String> args = new LinkedList<>();
	args.add("--instance-name");
	args.add(config.getInstanceName());
	args.add("--password");
	args.add(config.getRootPassword());

	final Process initProcess = miniAccumulo.exec(
			Initialize.class,
			jvmArgs,
			args.toArray(new String[0]));

	cleanup.add(initProcess);

	final int ret = initProcess.waitFor();
	if (ret != 0) {
		throw new RuntimeException(
				"Initialize process returned " + ret + ". Check the logs in " + config.getLogDir() + " for errors.");
	}

	LOGGER.info("Starting MAC against instance " + config.getInstanceName() + " and zookeeper(s)  "
			+ config.getZooKeepers());

	for (int i = 0; i < config.getNumTservers(); i++) {
		cleanup.add(miniAccumulo.exec(
				TabletServer.class,
				jvmArgs));
	}

	cleanup.add(miniAccumulo.exec(
			Master.class,
			jvmArgs));
	cleanup.add(miniAccumulo.exec(
			SimpleGarbageCollector.class,
			jvmArgs));
}
 

import org.apache.accumulo.core.conf.Property; //导入依赖的package包/类
@Test
public void testConcurrency() throws Exception {

  conn.tableOperations().setProperty(table, Property.TABLE_MAJC_RATIO.getKey(), "1");
  conn.tableOperations().setProperty(table, Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");

  int numAccounts = 5000;

  TreeSet<Text> splits = new TreeSet<>();
  splits.add(new Text(fmtAcct(numAccounts / 4)));
  splits.add(new Text(fmtAcct(numAccounts / 2)));
  splits.add(new Text(fmtAcct(3 * numAccounts / 4)));

  conn.tableOperations().addSplits(table, splits);

  AtomicBoolean runFlag = new AtomicBoolean(true);

  populate(env, numAccounts);

  Random rand = new Random();
  Environment tenv = env;

  if (rand.nextBoolean()) {
    tenv = new FaultyConfig(env, (rand.nextDouble() * .4) + .1, .50);
  }

  List<Thread> threads = startTransfers(tenv, numAccounts, 20, runFlag);

  runVerifier(env, numAccounts, 100);

  runFlag.set(false);

  for (Thread thread : threads) {
    thread.join();
  }

  log.debug("txCount : " + txCount.get());
  Assert.assertTrue("txCount : " + txCount.get(), txCount.get() > 0);

  runVerifier(env, numAccounts, 1);
}