import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testWriteConsistencyLevel() {
    int levelsChecked = 0;

    // Test whether CassandraTransaction honors the write consistency level option
    for (CLevel writeLevel : CLevel.values()) {
        StandardBaseTransactionConfig.Builder b = new StandardBaseTransactionConfig.Builder();
        ModifiableConfiguration mc = GraphDatabaseConfiguration.buildConfiguration();
        mc.set(CASSANDRA_WRITE_CONSISTENCY, writeLevel.name());
        b.customOptions(mc);
        b.timestampProvider(Timestamps.MICRO);
        CassandraTransaction ct = new CassandraTransaction(b.build());
        assertEquals(writeLevel, ct.getWriteConsistencyLevel());
        levelsChecked++;
    }

    // Sanity check: if CLevel.values was empty, something is wrong with the test
    Preconditions.checkState(0 < levelsChecked);
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testReadConsistencyLevel() {
    int levelsChecked = 0;

    // Test whether CassandraTransaction honors the write consistency level option
    for (CLevel writeLevel : CLevel.values()) {
        StandardBaseTransactionConfig.Builder b = new StandardBaseTransactionConfig.Builder();
        ModifiableConfiguration mc = GraphDatabaseConfiguration.buildGraphConfiguration();
        mc.set(CASSANDRA_READ_CONSISTENCY, writeLevel.name());
        b.timestampProvider(TimestampProviders.MICRO);
        b.customOptions(mc);
        CassandraTransaction ct = new CassandraTransaction(b.build());
        assertEquals(writeLevel, ct.getReadConsistencyLevel());
        levelsChecked++;
    }

    // Sanity check: if CLevel.values was empty, something is wrong with the test
    Preconditions.checkState(0 < levelsChecked);
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testLocalNodeUsingExt() throws BackendException, InterruptedException {

    String baseDir = Joiner.on(File.separator).join("target", "es", "jvmlocal_ext");

    assertFalse(new File(baseDir + File.separator + "data").exists());

    CommonsConfiguration cc = new CommonsConfiguration(new BaseConfiguration());
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.data", "true");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.client", "false");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.local", "true");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.data", baseDir + File.separator + "data");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.work", baseDir + File.separator + "work");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.logs", baseDir + File.separator + "logs");
    ModifiableConfiguration config =
            new ModifiableConfiguration(GraphDatabaseConfiguration.ROOT_NS,
                    cc, BasicConfiguration.Restriction.NONE);
    config.set(INTERFACE, ElasticSearchSetup.NODE.toString(), INDEX_NAME);
    Configuration indexConfig = config.restrictTo(INDEX_NAME);
    IndexProvider idx = new ElasticSearchIndex(indexConfig);
    simpleWriteAndQuery(idx);
    idx.close();

    assertTrue(new File(baseDir + File.separator + "data").exists());
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testKeybasedGraphPartitioning() {
    Object[] options = {option(GraphDatabaseConfiguration.IDS_FLUSH), false,
                        option(VertexIDAssigner.PLACEMENT_STRATEGY), PropertyPlacementStrategy.class.getName(),
            option(PropertyPlacementStrategy.PARTITION_KEY), "clusterId"};
    clopen(options);

    int[] groupDegrees = {5,5,5,5,5,5,5,5};
    int numVertices = setupGroupClusters(groupDegrees,CommitMode.PER_VERTEX);

    IntSet partitionIds = new IntHashSet(numVertices); //to track the "spread" of partition ids
    for (int i=0;i<groupDegrees.length;i++) {
        TitanVertex g = getOnlyVertex(tx.query().has("groupid","group"+i));
        int partitionId = -1;
        for (TitanVertex v : g.query().direction(Direction.IN).labels("member").vertices()) {
            if (partitionId<0) partitionId = getPartitionID(v);
            assertEquals(partitionId,getPartitionID(v));
            partitionIds.add(partitionId);
        }
    }
    assertTrue(partitionIds.size()>numPartitions/2); //This is a probabilistic test that might fail
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testLocalNodeUsingExt() throws BackendException, InterruptedException {

    String baseDir = Joiner.on("/").join("target", "es", "jvmlocal_ext");

    assertFalse(new File(baseDir + File.separator + "data").exists());

    CommonsConfiguration cc = new CommonsConfiguration(new BaseConfiguration());
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.data", "true");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.client", "false");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.local", "true");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.data", baseDir + File.separator + "data");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.work", baseDir + File.separator + "work");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.path.logs", baseDir + File.separator + "logs");
    ModifiableConfiguration config =
            new ModifiableConfiguration(GraphDatabaseConfiguration.ROOT_NS,
                    cc, BasicConfiguration.Restriction.NONE);
    config.set(INTERFACE, ElasticSearchSetup.NODE, INDEX_NAME);
    Configuration indexConfig = config.restrictTo(INDEX_NAME);
    IndexProvider idx = new ElasticSearchIndex(indexConfig);
    simpleWriteAndQuery(idx);
    idx.close();

    assertTrue(new File(baseDir + File.separator + "data").exists());
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public StoreFeatures getFeatures() {

    Configuration c = GraphDatabaseConfiguration.buildGraphConfiguration();

    StandardStoreFeatures.Builder fb = new StandardStoreFeatures.Builder()
            .orderedScan(true).unorderedScan(true).batchMutation(true)
            .multiQuery(true).distributed(true).keyOrdered(true).storeTTL(true)
            .timestamps(true).preferredTimestamps(PREFERRED_TIMESTAMPS)
            .keyConsistent(c);

    try {
        fb.localKeyPartition(getDeployment() == Deployment.LOCAL);
    } catch (Exception e) {
        logger.warn("Unexpected exception during getDeployment()", e);
    }

    return fb.build();
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public WriteConfiguration getConfig() {
    final KeyColumnValueStoreManager manager = new InMemoryStoreManager(Configuration.EMPTY);
    ModifiableConfiguration config = GraphDatabaseConfiguration.buildGraphConfiguration();
    config.set(GraphDatabaseConfiguration.TIMESTAMP_PROVIDER, TimestampProviders.MICRO);
    try {
        return new KCVSConfiguration(new BackendOperation.TransactionalProvider() {
            @Override
            public StoreTransaction openTx() throws BackendException {
                return manager.beginTransaction(StandardBaseTransactionConfig.of(TimestampProviders.MICRO, manager.getFeatures().getKeyConsistentTxConfig()));
            }

            @Override
            public void close() throws BackendException {
                manager.close();
            }
        }, config, manager.openDatabase("titan"),"general");
    } catch (BackendException e) {
        throw new RuntimeException(e);
    }
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@After
public void verifyMocks() {
    ctrl.verify();
    ctrl.reset();

    // Check capture created in the @Before method
    assertTrue(txConfigCapture.hasCaptured());
    List<BaseTransactionConfig> txCfgs = txConfigCapture.getValues();
    assertEquals(2, txCfgs.size());
    // First backing store transaction should use default tx config
    assertEquals("default", txCfgs.get(0).getCustomOption(GraphDatabaseConfiguration.UNIQUE_INSTANCE_ID));
    // Second backing store transaction should use global strong consistency config
    assertEquals("global",  txCfgs.get(1).getCustomOption(GraphDatabaseConfiguration.UNIQUE_INSTANCE_ID));
    // The order in which these transactions are opened isn't really significant;
    // testing them in order is kind of overspecifying the impl's behavior.
    // Could probably relax the ordering selectively here with some thought, but
    // I want to keep order checking on in general for the EasyMock control.
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public void clopen(Object... settings) {
    if (settings!=null && settings.length>0) {
        if (graph!=null && graph.isOpen()) {
            Preconditions.checkArgument(!graph.vertices().hasNext() &&
                !graph.edges().hasNext(),"Graph cannot be re-initialized for InMemory since that would delete all data");
            graph.close();
        }
        Map<TestConfigOption,Object> options = validateConfigOptions(settings);
        ModifiableConfiguration config = GraphDatabaseConfiguration.buildGraphConfiguration();
        config.set(GraphDatabaseConfiguration.STORAGE_BACKEND,"inmemory");
        for (Map.Entry<TestConfigOption,Object> option : options.entrySet()) {
            config.set(option.getKey().option, option.getValue(), option.getKey().umbrella);
        }
        open(config.getConfiguration());
    }
    newTx();
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
/**
 * Constructs a new TitanTransaction configuration with default configuration parameters.
 */
public StandardTransactionBuilder(GraphDatabaseConfiguration graphConfig, StandardTitanGraph graph) {
    Preconditions.checkNotNull(graphConfig);
    Preconditions.checkNotNull(graph);
    if (graphConfig.isReadOnly()) readOnly();
    if (graphConfig.isBatchLoading()) enableBatchLoading();
    this.graph = graph;
    this.defaultSchemaMaker = graphConfig.getDefaultSchemaMaker();
    this.assignIDsImmediately = graphConfig.hasFlushIDs();
    this.forceIndexUsage = graphConfig.hasForceIndexUsage();
    this.groupName = graphConfig.getMetricsPrefix();

    // KAFKA PRODUCER
    boolean logAll = graphConfig.getLogAllTransactions();

    if (logAll)
        this.logIdentifier = graphConfig.getAllLogTransactionName();
    else
        this.logIdentifier = null;

    this.propertyPrefetching = graphConfig.hasPropertyPrefetching();
    this.writableCustomOptions = GraphDatabaseConfiguration.buildGraphConfiguration();
    this.customOptions = new MergedConfiguration(writableCustomOptions, graphConfig.getConfiguration());
    vertexCacheSize(graphConfig.getTxVertexCacheSize());
    dirtyVertexSize(graphConfig.getTxDirtyVertexSize());
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Test
public void testLocalNodeUsingExtAndIndexDirectory() throws BackendException, InterruptedException {

    String baseDir = Joiner.on(File.separator).join("target", "es", "jvmlocal_ext2");

    assertFalse(new File(baseDir + File.separator + "data").exists());

    CommonsConfiguration cc = new CommonsConfiguration(new BaseConfiguration());
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.data", "true");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.client", "false");
    cc.set("index." + INDEX_NAME + ".elasticsearch.ext.node.local", "true");
    ModifiableConfiguration config =
            new ModifiableConfiguration(GraphDatabaseConfiguration.ROOT_NS,
                    cc, BasicConfiguration.Restriction.NONE);
    config.set(INTERFACE, ElasticSearchSetup.NODE.toString(), INDEX_NAME);
    config.set(INDEX_DIRECTORY, baseDir, INDEX_NAME);
    Configuration indexConfig = config.restrictTo(INDEX_NAME);
    IndexProvider idx = new ElasticSearchIndex(indexConfig);
    simpleWriteAndQuery(idx);
    idx.close();

    assertTrue(new File(baseDir + File.separator + "data").exists());
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
public VertexIDAssignerTest(Boolean partition, int partitionMax, int[] localPartition) {
    MockIDAuthority idAuthority = new MockIDAuthority(11, partitionMax);

    StandardStoreFeatures.Builder fb = new StandardStoreFeatures.Builder();

    if (null != localPartition) {
        fb.localKeyPartition(true);
        idAuthority.setLocalPartition(localPartition);
    }
    StoreFeatures features = fb.build();

    ModifiableConfiguration config = GraphDatabaseConfiguration.buildConfiguration();
    config.set(GraphDatabaseConfiguration.CLUSTER_PARTITION,partition);
    config.set(GraphDatabaseConfiguration.CLUSTER_MAX_PARTITIONS,1024);
    idAssigner = new VertexIDAssigner(config, idAuthority, features);
    System.out.println("Partition: " + partition);
    System.out.println("partitionMax: " + partitionMax);
    System.out.println("localPartition: " + Arrays.toString(localPartition));
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
private synchronized void shutdownInternal() throws TitanException {
    if (!isOpen) return;
    try {
        //Unregister instance
        ModifiableConfiguration globalConfig = GraphDatabaseConfiguration.getGlobalSystemConfig(backend);
        globalConfig.remove(REGISTRATION_TIME,config.getUniqueGraphId());

        super.shutdown();
        IOUtils.closeQuietly(serializer);
        idAssigner.close();
        backend.close();
        queryCache.close();
    } catch (BackendException e) {
        throw new TitanException("Could not close storage backend", e);
    } finally {
        isOpen = false;
    }
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public void setConf(final Configuration config) {
    super.setConf(config);

    // Copy some Titan configuration keys to the Hadoop Configuration keys used by Cassandra's ColumnFamilyInputFormat
    ConfigHelper.setInputInitialAddress(config, titanConf.get(GraphDatabaseConfiguration.STORAGE_HOSTS)[0]);
    if (titanConf.has(GraphDatabaseConfiguration.STORAGE_PORT))
        ConfigHelper.setInputRpcPort(config, String.valueOf(titanConf.get(GraphDatabaseConfiguration.STORAGE_PORT)));
    if (titanConf.has(GraphDatabaseConfiguration.AUTH_USERNAME))
        ConfigHelper.setInputKeyspaceUserName(config, titanConf.get(GraphDatabaseConfiguration.AUTH_USERNAME));
    if (titanConf.has(GraphDatabaseConfiguration.AUTH_PASSWORD))
        ConfigHelper.setInputKeyspacePassword(config, titanConf.get(GraphDatabaseConfiguration.AUTH_PASSWORD));

    // Copy keyspace, force the CF setting to edgestore, honor widerows when set
    final boolean wideRows = config.getBoolean(INPUT_WIDEROWS_CONFIG, false);
    // Use the setInputColumnFamily overload that includes a widerows argument; using the overload without this argument forces it false
    ConfigHelper.setInputColumnFamily(config, titanConf.get(AbstractCassandraStoreManager.CASSANDRA_KEYSPACE),
            mrConf.get(TitanHadoopConfiguration.COLUMN_FAMILY_NAME), wideRows);
    log.debug("Set keyspace: {}", titanConf.get(AbstractCassandraStoreManager.CASSANDRA_KEYSPACE));

    // Set the column slice bounds via Faunus's vertex query filter
    final SlicePredicate predicate = new SlicePredicate();
    final int rangeBatchSize = config.getInt(RANGE_BATCH_SIZE_CONFIG, Integer.MAX_VALUE);
    predicate.setSlice_range(getSliceRange(TitanHadoopSetupCommon.DEFAULT_SLICE_QUERY, rangeBatchSize)); // TODO stop slicing the whole row
    ConfigHelper.setInputSlicePredicate(config, predicate);
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
public InMemoryStoreManager(final Configuration configuration) {

        stores = new ConcurrentHashMap<String, InMemoryKeyColumnValueStore>();

        features = new StandardStoreFeatures.Builder()
            .orderedScan(true)
            .unorderedScan(true)
            .keyOrdered(true)
            .persists(false)
            .keyConsistent(GraphDatabaseConfiguration.buildGraphConfiguration())
            .build();

//        features = new StoreFeatures();
//        features.supportsOrderedScan = true;
//        features.supportsUnorderedScan = true;
//        features.supportsBatchMutation = false;
//        features.supportsTxIsolation = false;
//        features.supportsConsistentKeyOperations = true;
//        features.supportsLocking = false;
//        features.isDistributed = false;
//        features.supportsMultiQuery = false;
//        features.isKeyOrdered = true;
//        features.hasLocalKeyPartition = false;
    }
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
/**
 * Constructs a new TitanTransaction configuration with default configuration parameters.
 */
public StandardTransactionBuilder(GraphDatabaseConfiguration graphConfig, StandardTitanGraph graph) {
    Preconditions.checkNotNull(graphConfig);
    Preconditions.checkNotNull(graph);
    if (graphConfig.isReadOnly()) readOnly();
    if (graphConfig.isBatchLoading()) enableBatchLoading();
    this.graph = graph;
    this.defaultSchemaMaker = graphConfig.getDefaultSchemaMaker();
    this.assignIDsImmediately = graphConfig.hasFlushIDs();
    this.forceIndexUsage = graphConfig.hasForceIndexUsage();
    this.groupName = graphConfig.getMetricsPrefix();
    this.propertyPrefetching = graphConfig.hasPropertyPrefetching();
    this.writableCustomOptions = GraphDatabaseConfiguration.buildGraphConfiguration();
    this.customOptions = new MergedConfiguration(writableCustomOptions, graphConfig.getConfiguration());
    vertexCacheSize(graphConfig.getTxVertexCacheSize());
    dirtyVertexSize(graphConfig.getTxDirtyVertexSize());

    // KAFKA PRODUCER
    this.logIdentifier = null;
    boolean logAll = graphConfig.getLogAllTransactions();

    if (logAll) {
        this.logIdentifier = graphConfig.getAllLogTransactionName();
    }
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
/**
 * Clears out the entire graph. This will delete ALL of the data stored in this graph and the data will NOT be
 * recoverable. This method is intended only for development and testing use.
 *
 * @param graph
 * @throws IllegalArgumentException if the graph has not been shut down
 * @throws com.thinkaurelius.titan.core.TitanException if clearing the storage is unsuccessful
 */
public static final void clear(TitanGraph graph) {
    Preconditions.checkNotNull(graph);
    Preconditions.checkArgument(graph instanceof StandardTitanGraph,"Invalid graph instance detected: %s",graph.getClass());
    StandardTitanGraph g = (StandardTitanGraph)graph;
    Preconditions.checkArgument(!g.isOpen(),"Graph needs to be shut down before it can be cleared.");
    final GraphDatabaseConfiguration config = g.getConfiguration();
    BackendOperation.execute(new Callable<Boolean>(){
        @Override
        public Boolean call() throws Exception {
            config.getBackend().clearStorage();
            return true;
        }
        @Override
        public String toString() { return "ClearBackend"; }
    },new StandardDuration(20, TimeUnit.SECONDS));
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public StoreFeatures getFeatures() {

    Configuration c = GraphDatabaseConfiguration.buildConfiguration();

    StandardStoreFeatures.Builder fb = new StandardStoreFeatures.Builder()
            .orderedScan(true).unorderedScan(true).batchMutation(true)
            .multiQuery(true).distributed(true).keyOrdered(true).storeTTL(true)
            .timestamps(true).preferredTimestamps(PREFERRED_TIMESTAMPS)
            .keyConsistent(c);

    try {
        fb.localKeyPartition(getDeployment() == Deployment.LOCAL);
    } catch (Exception e) {
        logger.warn("Unexpected exception during getDeployment()", e);
    }

    return fb.build();
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
/**
 * Clears out the entire graph. This will delete ALL of the data stored in this graph and the data will NOT be
 * recoverable. This method is intended only for development and testing use.
 *
 * @param graph
 * @throws IllegalArgumentException if the graph has not been shut down
 * @throws com.thinkaurelius.titan.core.TitanException if clearing the storage is unsuccessful
 */
public static final void clear(TitanGraph graph) {
    Preconditions.checkNotNull(graph);
    Preconditions.checkArgument(graph instanceof StandardTitanGraph,"Invalid graph instance detected: %s",graph.getClass());
    StandardTitanGraph g = (StandardTitanGraph)graph;
    Preconditions.checkArgument(!g.isOpen(),"Graph needs to be shut down before it can be cleared.");
    final GraphDatabaseConfiguration config = g.getConfiguration();
    BackendOperation.execute(new Callable<Boolean>(){
        @Override
        public Boolean call() throws Exception {
            config.getBackend().clearStorage();
            return true;
        }
        @Override
        public String toString() { return "ClearBackend"; }
    }, Duration.ofSeconds(20));
}
 

import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration; //导入依赖的package包/类
@Override
public StoreFeatures getFeatures() {

    Configuration c = GraphDatabaseConfiguration.buildConfiguration();

    StandardStoreFeatures.Builder fb = new StandardStoreFeatures.Builder()
            .orderedScan(true).unorderedScan(true).batchMutation(true)
            .multiQuery(true).distributed(true).keyOrdered(true).storeTTL(true)
            .timestamps(true).preferredTimestamps(PREFERRED_TIMESTAMPS)
            .locking(true)
            .keyConsistent(c);

    try {
        fb.localKeyPartition(getDeployment() == Deployment.LOCAL);
    } catch (Exception e) {
        logger.warn("Unexpected exception during getDeployment()", e);
    }

    return fb.build();
}