import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public UpdateParameters(CFMetaData metadata,
                        PartitionColumns updatedColumns,
                        QueryOptions options,
                        long timestamp,
                        int ttl,
                        Map<DecoratedKey, Partition> prefetchedRows)
throws InvalidRequestException
{
    this.metadata = metadata;
    this.updatedColumns = updatedColumns;
    this.options = options;

    this.nowInSec = FBUtilities.nowInSeconds();
    this.timestamp = timestamp;
    this.ttl = ttl;

    this.deletionTime = new DeletionTime(timestamp, nowInSec);

    this.prefetchedRows = prefetchedRows;

    // We use MIN_VALUE internally to mean the absence of of timestamp (in Selection, in sstable stats, ...), so exclude
    // it to avoid potential confusion.
    if (timestamp == Long.MIN_VALUE)
        throw new InvalidRequestException(String.format("Out of bound timestamp, must be in [%d, %d]", Long.MIN_VALUE + 1, Long.MAX_VALUE));
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
private static ByteBuffer maybeTakeFromPool(int size, boolean allocateOnHeapWhenExhausted)
{
    if (size < 0)
        throw new IllegalArgumentException("Size must be positive (" + size + ")");

    if (size == 0)
        return EMPTY_BUFFER;

    if (size > CHUNK_SIZE)
    {
        if (logger.isTraceEnabled())
            logger.trace("Requested buffer size {} is bigger than {}, allocating directly",
                         FBUtilities.prettyPrintMemory(size),
                         FBUtilities.prettyPrintMemory(CHUNK_SIZE));

        return localPool.get().allocate(size, allocateOnHeapWhenExhausted);
    }

    return localPool.get().get(size);
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
private static UnfilteredRowMergeIterator create(List<UnfilteredRowIterator> iterators, int nowInSec, MergeListener listener)
{
    try
    {
        //checkForInvalidInput(iterators);
        return new UnfilteredRowMergeIterator(iterators.get(0).metadata(),
                                              iterators,
                                              collectColumns(iterators),
                                              collectPartitionLevelDeletion(iterators, listener),
                                              nowInSec,
                                              iterators.get(0).isReverseOrder(),
                                              listener);
    }
    catch (RuntimeException | Error e)
    {
        try
        {
            FBUtilities.closeAll(iterators);
        }
        catch (Exception suppressed)
        {
            e.addSuppressed(suppressed);
        }
        throw e;
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
private String getLocation()
{
    Collection<InetAddress> localAddresses = FBUtilities.getAllLocalAddresses();

    for (InetAddress address : localAddresses)
    {
        for (String location : split.getLocations())
        {
            InetAddress locationAddress = null;
            try
            {
                locationAddress = InetAddress.getByName(location);
            }
            catch (UnknownHostException e)
            {
                throw new AssertionError(e);
            }
            if (address.equals(locationAddress))
            {
                return location;
            }
        }
    }
    return split.getLocations()[0];
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public static void validateDeletion(CFMetaData metadata, Deletion del) throws org.apache.cassandra.exceptions.InvalidRequestException
{

    if (del.super_column != null)
        validateColumnNames(metadata, (ByteBuffer)null, Arrays.asList(del.super_column));

    if (del.predicate != null)
        validateSlicePredicate(metadata, del.super_column, del.predicate);

    if (!metadata.isSuper() && del.super_column != null)
    {
        String msg = String.format("Deletion of super columns is not possible on a standard table (KeySpace=%s Table=%s Deletion=%s)", metadata.ksName, metadata.cfName, del);
        throw new org.apache.cassandra.exceptions.InvalidRequestException(msg);
    }

    if (metadata.isCounter())
    {
        // forcing server timestamp even if a timestamp was set for coherence with other counter operation
        del.timestamp = FBUtilities.timestampMicros();
    }
    else if (!del.isSetTimestamp())
    {
        throw new org.apache.cassandra.exceptions.InvalidRequestException("Deletion timestamp is not optional for non commutative table " + metadata.cfName);
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public NetworkTopologyStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions) throws ConfigurationException
{
    super(keyspaceName, tokenMetadata, snitch, configOptions);
    this.snitch = snitch;

    Map<String, Integer> newDatacenters = new HashMap<String, Integer>();
    if (configOptions != null)
    {
        for (Entry<String, String> entry : configOptions.entrySet())
        {
            String dc = entry.getKey();
            if (dc.equalsIgnoreCase("replication_factor"))
                throw new ConfigurationException("replication_factor is an option for SimpleStrategy, not NetworkTopologyStrategy");
            Integer replicas = Integer.valueOf(entry.getValue());
            newDatacenters.put(dc, replicas);
        }
    }

    datacenters = Collections.unmodifiableMap(newDatacenters);
    logger.trace("Configured datacenter replicas are {}", FBUtilities.toString(datacenters));
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
@Test
public void testGetNeighborsPlusOneInLocalDC() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();
    
    // generate rf+1 nodes, and ensure that all nodes are returned
    Set<InetAddress> expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    expected.remove(FBUtilities.getBroadcastAddress());
    // remove remote endpoints
    TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
    HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
    expected = Sets.intersection(expected, localEndpoints);

    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
    Set<InetAddress> neighbors = new HashSet<InetAddress>();
    for (Range<Token> range : ranges)
    {
        neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
    }
    assertEquals(expected, neighbors);
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
/**
 * Does a full, blocking rebuild of the indexes specified by columns from the sstables.
 * Does nothing if columns is empty.
 *
 * Caller must acquire and release references to the sstables used here.
 *
 * @param sstables the data to build from
 * @param idxNames the list of columns to index, ordered by comparator
 */
public void maybeBuildSecondaryIndexes(Collection<SSTableReader> sstables, Set<String> idxNames)
{
    if (idxNames.isEmpty())
        return;

    logger.info(String.format("Submitting index build of %s for data in %s",
                              idxNames, StringUtils.join(sstables, ", ")));

    SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs, idxNames, new ReducingKeyIterator(sstables));
    Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
    FBUtilities.waitOnFuture(future);

    flushIndexesBlocking();

    logger.info("Index build of {} complete", idxNames);
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
/**
 * be careful about just blindly updating ApplicationState.INTERNAL_IP everytime we read the yaml file,
 * as that can cause connections to get unnecessarily reset (via IESCS.onChange()).
 */
private void maybeSetApplicationState()
{
    if (localNodeData.dcLocalAddress == null)
        return;
    final EndpointState es = Gossiper.instance.getEndpointStateForEndpoint(FBUtilities.getBroadcastAddress());
    if (es == null)
        return;
    final VersionedValue vv = es.getApplicationState(ApplicationState.INTERNAL_IP);
    if ((vv != null && !vv.value.equals(localNodeData.dcLocalAddress.getHostAddress()))
        || vv == null)
    {
        Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP,
            StorageService.instance.valueFactory.internalIP(localNodeData.dcLocalAddress.getHostAddress()));
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
CqlRecordWriter(Configuration conf)
{
    this.conf = conf;
    this.queueSize = conf.getInt(CqlOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
    batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32);
    this.clients = new HashMap<>();
    String keyspace = ConfigHelper.getOutputKeyspace(conf);

    try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf))
    {
        Metadata metadata = cluster.getMetadata();
        ringCache = new NativeRingCache(conf, metadata);
        TableMetadata tableMetadata = metadata.getKeyspace(Metadata.quote(keyspace)).getTable(ConfigHelper.getOutputColumnFamily(conf));
        clusterColumns = tableMetadata.getClusteringColumns();
        partitionKeyColumns = tableMetadata.getPartitionKey();

        String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
        if (cqlQuery.toLowerCase().startsWith("insert"))
            throw new UnsupportedOperationException("INSERT with CqlRecordWriter is not supported, please use UPDATE/DELETE statement");
        cql = appendKeyWhereClauses(cqlQuery);
    }
    catch (Exception e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
@Test
public void testGetColumn()
{
    Keyspace keyspace = Keyspace.open("Keyspace1");
    CellNameType type = keyspace.getColumnFamilyStore("Standard1").getComparator();
    Mutation rm;
    DecoratedKey dk = Util.dk("key1");

    // add data
    rm = new Mutation("Keyspace1", dk.getKey());
    rm.add("Standard1", Util.cellname("Column1"), ByteBufferUtil.bytes("abcd"), 0);
    rm.apply();

    ReadCommand command = new SliceByNamesReadCommand("Keyspace1", dk.getKey(), "Standard1", System.currentTimeMillis(), new NamesQueryFilter(FBUtilities.singleton(Util.cellname("Column1"), type)));
    Row row = command.getRow(keyspace);
    Cell col = row.cf.getColumn(Util.cellname("Column1"));
    assertEquals(col.value(), ByteBuffer.wrap("abcd".getBytes()));
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public PropertyFileSnitch() throws ConfigurationException
{
    reloadConfiguration();

    try
    {
        FBUtilities.resourceToFile(SNITCH_PROPERTIES_FILENAME);
        Runnable runnable = new WrappedRunnable()
        {
            protected void runMayThrow() throws ConfigurationException
            {
                reloadConfiguration();
            }
        };
        ResourceWatcher.watch(SNITCH_PROPERTIES_FILENAME, runnable, 60 * 1000);
    }
    catch (ConfigurationException ex)
    {
        logger.error("{} found, but does not look like a plain file. Will not watch it for changes", SNITCH_PROPERTIES_FILENAME);
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
/**
 * Send merkle tree request to every involved neighbor.
 */
public void sendTreeRequests(Collection<InetAddress> endpoints)
{
    // send requests to all nodes
    List<InetAddress> allEndpoints = new ArrayList<>(endpoints);
    allEndpoints.add(FBUtilities.getBroadcastAddress());

    if (isSequential)
        makeSnapshots(endpoints);

    this.gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(System.currentTimeMillis());

    for (InetAddress endpoint : allEndpoints)
        treeRequests.add(endpoint);

    logger.info(String.format("[repair #%s] requesting merkle trees for %s (to %s)", desc.sessionId, desc.columnFamily, allEndpoints));
    treeRequests.start();
    requestsSent.signalAll();
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public static void validateKey(CFMetaData metadata, ByteBuffer key) throws org.apache.cassandra.exceptions.InvalidRequestException
{
    if (key == null || key.remaining() == 0)
    {
        throw new org.apache.cassandra.exceptions.InvalidRequestException("Key may not be empty");
    }

    // check that key can be handled by FBUtilities.writeShortByteArray
    if (key.remaining() > FBUtilities.MAX_UNSIGNED_SHORT)
    {
        throw new org.apache.cassandra.exceptions.InvalidRequestException("Key length of " + key.remaining() +
                                                                          " is longer than maximum of " +
                                                                          FBUtilities.MAX_UNSIGNED_SHORT);
    }

    try
    {
        metadata.getKeyValidator().validate(key);
    }
    catch (MarshalException e)
    {
        throw new org.apache.cassandra.exceptions.InvalidRequestException(e.getMessage());
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
public Future<Pair<KeyCacheKey, RowIndexEntry>> deserialize(DataInputStream input, ColumnFamilyStore cfs) throws IOException
{
    int keyLength = input.readInt();
    if (keyLength > FBUtilities.MAX_UNSIGNED_SHORT)
    {
        throw new IOException(String.format("Corrupted key cache. Key length of %d is longer than maximum of %d",
                                            keyLength, FBUtilities.MAX_UNSIGNED_SHORT));
    }
    ByteBuffer key = ByteBufferUtil.read(input, keyLength);
    int generation = input.readInt();
    SSTableReader reader = findDesc(generation, cfs.getSSTables());
    input.readBoolean(); // backwards compatibility for "promoted indexes" boolean
    if (reader == null)
    {
        RowIndexEntry.Serializer.skipPromotedIndex(input);
        return null;
    }
    RowIndexEntry entry = reader.metadata.comparator.rowIndexEntrySerializer().deserialize(input, reader.descriptor.version);
    return Futures.immediateFuture(Pair.create(new KeyCacheKey(cfs.metadata.cfId, reader.descriptor, key), entry));
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
/**
 * Return the rack for which an endpoint resides in
 *
 * @param endpoint the endpoint to process
 * @return string of rack
 */
public String getRack(InetAddress endpoint)
{
    if (endpoint.equals(FBUtilities.getBroadcastAddress()))
        return myRack;

    EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
    if (epState == null || epState.getApplicationState(ApplicationState.RACK) == null)
    {
        if (psnitch == null)
        {
            if (savedEndpoints == null)
                savedEndpoints = SystemKeyspace.loadDcRackInfo();
            if (savedEndpoints.containsKey(endpoint))
                return savedEndpoints.get(endpoint).get("rack");
            return DEFAULT_RACK;
        }
        else
            return psnitch.getRack(endpoint);
    }
    return epState.getApplicationState(ApplicationState.RACK).value;
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
/**
 * Starts sending/receiving our list of differences to/from the remote endpoint: creates a callback
 * that will be called out of band once the streams complete.
 */
protected void startSync(List<Range<Token>> differences)
{
    InetAddress local = FBUtilities.getBroadcastAddress();
    // We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
    InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
    InetAddress preferred = SystemKeyspace.getPreferredIP(dst);

    String message = String.format("Performing streaming repair of %d ranges with %s", differences.size(), dst);
    logger.info("[repair #{}] {}", desc.sessionId, message);
    boolean isIncremental = false;
    if (desc.parentSessionId != null)
    {
        ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId);
        isIncremental = prs.isIncremental;
    }
    Tracing.traceRepair(message);
    new StreamPlan("Repair", repairedAt, 1, false, isIncremental).listeners(this)
                                        .flushBeforeTransfer(true)
                                        // request ranges from the remote node
                                        .requestRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        // send ranges to the remote node
                                        .transferRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily)
                                        .execute();
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
@Test
public void testSnapshotMultipleRepairs()
{
    ColumnFamilyStore store = prepareColumnFamilyStore();
    Set<SSTableReader> original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables);
    UUID prsId = UUID.randomUUID();
    ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true);
    ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.cfId, prsId);

    UUID prsId2 = UUID.randomUUID();
    ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true);
    boolean exception = false;
    try
    {
        ActiveRepairService.instance.getParentRepairSession(prsId2).maybeSnapshot(store.metadata.cfId, prsId2);
    }
    catch (Throwable t)
    {
        exception = true;
    }
    assertTrue(exception);
    try (Refs<SSTableReader> refs = ActiveRepairService.instance.getParentRepairSession(prsId).getActiveRepairedSSTableRefsForAntiCompaction(store.metadata.cfId, prsId))
    {
        assertEquals(original, Sets.newHashSet(refs.iterator()));
    }
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
@Test
public void testGetNeighborsTimesTwoInSpecifiedHosts() throws Throwable
{
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    // generate rf*2 nodes, and ensure that only neighbors specified by the hosts are returned
    addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
    AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
    List<InetAddress> expected = new ArrayList<>();
    for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
    {
        expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
    }

    expected.remove(FBUtilities.getBroadcastAddress());
    Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName());

   assertEquals(expected.get(0), ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts).iterator().next());
}
 

import org.apache.cassandra.utils.FBUtilities; //导入依赖的package包/类
private static Mutation createLegacyHint(Mutation mutation, long now, UUID targetId)
{
    int version = MessagingService.VERSION_21;
    int ttl = mutation.smallestGCGS();
    UUID hintId = UUIDGen.getTimeUUID();

    ByteBuffer key = UUIDType.instance.decompose(targetId);
    Clustering clustering = SystemKeyspace.LegacyHints.comparator.make(hintId, version);
    ByteBuffer value = ByteBuffer.wrap(FBUtilities.serialize(mutation, Mutation.serializer, version));
    Cell cell = BufferCell.expiring(SystemKeyspace.LegacyHints.compactValueColumn(),
                                    now,
                                    ttl,
                                    FBUtilities.nowInSeconds(),
                                    value);
    return new Mutation(PartitionUpdate.singleRowUpdate(SystemKeyspace.LegacyHints,
                                                        key,
                                                        BTreeRow.singleCellRow(clustering, cell)));
}