import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
@Override
public Placement newPlacement(String placement) throws ConnectionException {
    String[] parsed = PlacementUtil.parsePlacement(placement);
    String keyspaceName = parsed[0];
    String cfPrefix = parsed[1];

    CassandraKeyspace keyspace = _keyspaceMap.get(keyspaceName);
    if (keyspace == null) {
        throw new UnknownPlacementException(format(
                "Placement string refers to unknown or non-local Cassandra keyspace: %s", keyspaceName), placement);
    }

    KeyspaceDefinition keyspaceDef = keyspace.getAstyanaxKeyspace().describeKeyspace();
    ColumnFamily<ByteBuffer,Composite> columnFamily = getColumnFamily(keyspaceDef, cfPrefix, "blob", placement,
            new SpecificCompositeSerializer(CompositeType.getInstance(Arrays.<AbstractType<?>>asList(
                    AsciiType.instance, IntegerType.instance))));

    return new BlobPlacement(placement, keyspace, columnFamily);
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
protected <C> ColumnFamily<ByteBuffer, C> getColumnFamily(KeyspaceDefinition keyspaceDef,
                                                          String prefix, String suffix, String placement,
                                                          Serializer<C> columnSerializer) throws IllegalArgumentException {
    // Create the column family object.  It must be keyed by a ByteBuffer because that's what
    // the AstyanaxTable.getRowKey() method returns.
    ColumnFamily<ByteBuffer, C> cf = new ColumnFamily<>(prefix + "_" + suffix,
            ByteBufferSerializer.get(), columnSerializer);

    // Verify that the column family exists in the Cassandra schema.
    ColumnFamilyDefinition cfDef = keyspaceDef.getColumnFamily(cf.getName());
    if (cfDef == null) {
        throw new UnknownPlacementException(format(
                "Placement string '%s' refers to unknown Cassandra %s column family in keyspace '%s': %s",
                placement, suffix, keyspaceDef.getName(), cf.getName()), placement);
    }
    return cf;
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
@Override
public Placement newPlacement(String placement) throws ConnectionException {
    String[] parsed = PlacementUtil.parsePlacement(placement);
    String keyspaceName = parsed[0];
    String cfPrefix = parsed[1];

    CassandraKeyspace keyspace = _keyspaceMap.get(keyspaceName);
    if (keyspace == null) {
        throw new UnknownPlacementException(format(
                "Placement string refers to unknown or non-local Cassandra keyspace: %s", keyspaceName), placement);
    }

    KeyspaceDefinition keyspaceDef = keyspace.getAstyanaxKeyspace().describeKeyspace();
    AnnotatedCompositeSerializer<DeltaKey> deltaKeySerializer  = new AnnotatedCompositeSerializer<DeltaKey>(DeltaKey.class);

    // DDL's are not actually configurable due to the way we abstract the names from the placements here.
    // In the future, we should either phase out the DDL config or change the implementation here to conform to it.
    ColumnFamily<ByteBuffer, UUID> deltaCf = getColumnFamily(keyspaceDef, cfPrefix, "delta", placement, TimeUUIDSerializer.get());
    ColumnFamily<ByteBuffer, DeltaKey> blockedDeltaCf = getColumnFamily(keyspaceDef, cfPrefix, "delta_v2", placement, deltaKeySerializer);
    ColumnFamily<ByteBuffer, UUID> auditCf = getColumnFamily(keyspaceDef, cfPrefix, "audit", placement, TimeUUIDSerializer.get());
    ColumnFamily<ByteBuffer, UUID> deltaHistoryCf = getColumnFamily(keyspaceDef, cfPrefix, "history", placement, TimeUUIDSerializer.get());

    // Calculate the data centers on demand since they may change in a live system.
    return new DeltaPlacement(placement, keyspace, deltaCf, blockedDeltaCf, auditCf, deltaHistoryCf);
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
public CassandraReplication(KeyspaceDefinition keyspaceDefinition) {
    _networkTopology = keyspaceDefinition.getStrategyClass().endsWith("NetworkTopologyStrategy");

    if (_networkTopology) {
        // This algorithm should match the NetworkTopologyStrategy.getReplicationFactor() method.
        // Strategy options is a Map of data center name -> replication factor.
        int replicationFactor = 0;
        ImmutableMap.Builder<String, Integer> dataCenterBuilder = ImmutableMap.builder();
        for (Map.Entry<String, String> option : keyspaceDefinition.getStrategyOptions().entrySet()) {
            String dataCenter = option.getKey();
            int repFactor = Integer.parseInt(option.getValue());
            replicationFactor += repFactor;
            dataCenterBuilder.put(dataCenter, repFactor);
        }
        _replicationFactor = replicationFactor;
        _replicationFactorByDataCenter = dataCenterBuilder.build();
    } else {
        // SimpleStrategy and OldNetworkTopologyStrategy both require a 'replication_factor' setting
        _replicationFactor = Integer.parseInt(keyspaceDefinition.getStrategyOptions().get("replication_factor"));
        _replicationFactorByDataCenter = ImmutableMap.of();
    }
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
@Override
public Map<String, String> getCompressionOptions(String cf) throws BackendException {
    try {
        Keyspace k = keyspaceContext.getClient();

        KeyspaceDefinition kdef = k.describeKeyspace();

        if (null == kdef) {
            throw new PermanentBackendException("Keyspace " + k.getKeyspaceName() + " is undefined");
        }

        ColumnFamilyDefinition cfdef = kdef.getColumnFamily(cf);

        if (null == cfdef) {
            throw new PermanentBackendException("Column family " + cf + " is undefined");
        }

        return cfdef.getCompressionOptions();
    } catch (ConnectionException e) {
        throw new PermanentBackendException(e);
    }
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
public void clearImpl() throws ConnectionException {
	Cluster cluster = columnFamilies.getCluster();
	String keyspaceName = columnFamilies.getKeyspaceName();
	if (log.isInfoEnabled())
		log.info("Clearing keyspace="+keyspaceName+" in cassandra");
	List<KeyspaceDefinition> keyspaces = cluster.describeKeyspaces();
	KeyspaceDefinition ourDef = null;
	for(KeyspaceDefinition kDef : keyspaces) {
		if(keyspaceName.equalsIgnoreCase(kDef.getName())) {
			ourDef = kDef;
			break;
		}
	}
	
	cluster.dropKeyspace(keyspaceName);
	OperationResult<SchemaChangeResult> result = cluster.addKeyspace(ourDef);
	
	columnFamilies.waitForNodesToBeUpToDate(result, 300000);
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private void createLocksKeyspace() throws ConnectionException {

        try {
            KeyspaceDefinition keyspaceDefinition = keyspace.describeKeyspace();
            if ( keyspaceDefinition != null ) {
                logger.info("Keyspace {} already exists", keyspace.getKeyspaceName());
                return;
            }
        } catch (ConnectionException ce){
            logger.debug( "Received a NotFoundException when attempting to describe keyspace.  It does not exist" );
        }

        ImmutableMap.Builder<String, Object> strategyOptions = getKeySpaceProps();
        ImmutableMap<String, Object> options =
            ImmutableMap.<String, Object>builder().put("strategy_class", cassandraFig.getLocksKeyspaceStrategy())
                .put("strategy_options", strategyOptions.build()).build();

        keyspace.createKeyspace(options);
        logger.info("Keyspace {} created with options {}",  keyspace.getKeyspaceName(), options.toString());

    }
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
/**
 * Check if the column family exists.  If it dosn't create it
 */
private void testAndCreateColumnFamilyDef( MultiTenantColumnFamilyDefinition columnFamily )
        throws ConnectionException {
    final KeyspaceDefinition keyspaceDefinition = keyspace.describeKeyspace();

    final ColumnFamilyDefinition existing =
            keyspaceDefinition.getColumnFamily( columnFamily.getColumnFamily().getName() );

    if ( existing != null ) {
        logger.info("Not creating columnfamily {}, it already exists.", columnFamily.getColumnFamily().getName());
        return;
    }

    keyspace.createColumnFamily( columnFamily.getColumnFamily(), columnFamily.getOptions() );

    // the CF def creation uses Asytanax, so manually check the schema agreement
    astyanaxWaitForSchemaAgreement();

    logger.info( "Created column family {}", columnFamily.getColumnFamily().getName() );

}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
@Override
public Collection<String> getKeyspacesForDataCenter(String cassandraDataCenter) {
    Set<String> keyspaces = Sets.newHashSet();
    for (Map.Entry<String, CassandraKeyspace> entry : _keyspaceMap.entrySet()) {
        KeyspaceDefinition keyspaceDefinition = describe(entry.getValue());
        if (replicatesTo(keyspaceDefinition, cassandraDataCenter)) {
            keyspaces.add(entry.getKey());
        }
    }
    return keyspaces;
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private boolean replicatesTo(KeyspaceDefinition keyspaceDefinition, String dataCenter) {
    if (keyspaceDefinition.getStrategyClass().endsWith("NetworkTopologyStrategy")) {
        String numReplicas = keyspaceDefinition.getStrategyOptions().get(dataCenter);
        return numReplicas != null && Integer.valueOf(numReplicas) != 0;
    }
    // Other strategies don't vary replication factor by data center.  Assume true.
    return true;
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private KeyspaceDefinition describe(CassandraKeyspace keyspace) {
    try {
        return keyspace.getAstyanaxKeyspace().describeKeyspace();
    } catch (ConnectionException e) {
        throw Throwables.propagate(e);
    }
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private MapStringToObject getKeyspaceOptions(KeyspaceDefinition keyspace) {
    MapStringToObject result = new MapStringToObject();
    for (FieldMetadata field : keyspace.getFieldsMetadata()) {
        result.put(field.getName(), keyspace.getFieldValue(field.getName()));
    }

    result.remove("CF_DEFS");
    return result;
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private ColumnFamily getLocksColumnFamily() {

        if ( columnFamily == null ) {

            columnFamily = ColumnFamily.newColumnFamily(
                CF_NAME, StringSerializer.get(), StringSerializer.get() );

            if ( logger.isDebugEnabled() ) {

                try {
                    final KeyspaceDefinition kd = keyspace.describeKeyspace();
                    final ColumnFamilyDefinition cfd = kd.getColumnFamily( columnFamily.getName() );
                    Map<String, Object> options = new HashMap<>( 1 );
                    options.put( "gc_grace_seconds", cfd.getGcGraceSeconds() );
                    options.put( "caching", cfd.getCaching() );
                    options.put( "compaction_strategy", cfd.getCompactionStrategy() );
                    options.put( "compaction_strategy_options", cfd.getCompactionStrategyOptions() );
                    logger.debug( "Locks column family {} exists with options: {}", cfd.getName(), options);

                } catch ( ConnectionException ce ) {
                    logger.warn("Error connecting to Cassandra for debug column family info", ce);
                }
            }
        }

        return columnFamily;
    }
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private void findExistingColumnFamilies(KeyspaceDefinition keySpaceMeta) {
	List<ColumnFamilyDefinition> cfList = keySpaceMeta.getColumnFamilyList();
	for(ColumnFamilyDefinition def : cfList) {
		loadColumnFamilyImpl(def);
	}
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
public static List<KeyspaceDefinition> getAllKeyspaces() throws ConnectionException {
	return cluster.describeKeyspaces();
}
 

import com.netflix.astyanax.ddl.KeyspaceDefinition; //导入依赖的package包/类
private ColumnFamily createLocksColumnFamily() throws ConnectionException {

        ColumnFamily<String, String> cflocks = ColumnFamily.newColumnFamily(
            CF_NAME, StringSerializer.get(), StringSerializer.get() );

        final KeyspaceDefinition kd = keyspace.describeKeyspace();
        final ColumnFamilyDefinition cfdef = kd.getColumnFamily( cflocks.getName() );

        if ( cfdef == null ) {

            // create only if does not already exist

            MultiTenantColumnFamilyDefinition mtcfd = new MultiTenantColumnFamilyDefinition(
                cflocks,
                BytesType.class.getSimpleName(),
                UTF8Type.class.getSimpleName(),
                BytesType.class.getSimpleName(),
                MultiTenantColumnFamilyDefinition.CacheOption.ALL
            );

            Map<String, Object> cfOptions = mtcfd.getOptions();

            // Additionally set the gc grace low
            cfOptions.put( "gc_grace_seconds", 60 );

            keyspace.createColumnFamily( mtcfd.getColumnFamily(), cfOptions );

            logger.info( "Created column family {}", mtcfd.getOptions() );

            cflocks = mtcfd.getColumnFamily();

        } else {
            return getLocksColumnFamily();
        }

        return cflocks;
    }