import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
private void grabCounterLocks(Keyspace keyspace, List<Lock> locks) throws WriteTimeoutException
{
    long startTime = System.nanoTime();

    for (Lock lock : LOCKS.bulkGet(getCounterLockKeys()))
    {
        long timeout = TimeUnit.MILLISECONDS.toNanos(getTimeout()) - (System.nanoTime() - startTime);
        try
        {
            if (!lock.tryLock(timeout, TimeUnit.NANOSECONDS))
                throw new WriteTimeoutException(WriteType.COUNTER, consistency(), 0, consistency().blockFor(keyspace));
            locks.add(lock);
        }
        catch (InterruptedException e)
        {
            throw new WriteTimeoutException(WriteType.COUNTER, consistency(), 0, consistency().blockFor(keyspace));
        }
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public void finish()
{
    for (int i = 0; i < replayHandlers.size(); i++)
    {
        ReplayWriteResponseHandler handler = replayHandlers.get(i);
        try
        {
            handler.get();
        }
        catch (WriteTimeoutException e)
        {
            logger.debug("Timed out replaying a batched mutation to a node, will write a hint");
            // writing hints for the rest to hints, starting from i
            writeHintsForUndeliveredEndpoints(i);
            return;
        }
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public static TimedOutException toThrift(RequestTimeoutException e)
{
    TimedOutException toe = new TimedOutException();
    if (e instanceof WriteTimeoutException)
    {
        WriteTimeoutException wte = (WriteTimeoutException)e;
        toe.setAcknowledged_by(wte.received);
        if (wte.writeType == WriteType.BATCH_LOG)
            toe.setAcknowledged_by_batchlog(false);
        else if (wte.writeType == WriteType.BATCH)
            toe.setAcknowledged_by_batchlog(true);
        else if (wte.writeType == WriteType.CAS)
            toe.setPaxos_in_progress(true);
    }
    return toe;
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testDuplicateCells() throws WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KS).getColumnFamilyStore(CF1);
    cfs.truncateBlocking();

    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 1L);
    cells.addCounter(cellname(1), 2L);
    cells.addCounter(cellname(1), 3L);
    cells.addCounter(cellname(1), 4L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();

    ColumnFamily current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    ByteBuffer context = current.getColumn(cellname(1)).value();
    assertEquals(10L, CounterContext.instance().total(context));
    assertEquals(ClockAndCount.create(1L, 10L), CounterContext.instance().getLocalClockAndCount(context));
    assertEquals(ClockAndCount.create(1L, 10L), cfs.getCachedCounter(bytes(1), cellname(1)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KS).getColumnFamilyStore(CF);
    CacheService.instance.invalidateCounterCache();

    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addColumn(new BufferCounterUpdateCell(cellname(1), 1L, FBUtilities.timestampMicros()));
    cells.addColumn(new BufferCounterUpdateCell(cellname(2), 2L, FBUtilities.timestampMicros()));
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    new CounterMutation(new Mutation(KS, bytes(2), cells), ConsistencyLevel.ONE).apply();

    // flush the counter cache and invalidate
    CacheService.instance.counterCache.submitWrite(Integer.MAX_VALUE).get();
    CacheService.instance.invalidateCounterCache();
    assertEquals(0, CacheService.instance.counterCache.size());

    // load from cache and validate
    CacheService.instance.counterCache.loadSaved(cfs);
    assertEquals(4, CacheService.instance.counterCache.size());
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(1), cellname(2)));
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(2), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(2), cellname(2)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
/**
 * Applies the counter mutation, returns the result Mutation (for replication to other nodes).
 *
 * 1. Grabs the striped cell-level locks in the proper order
 * 2. Gets the current values of the counters-to-be-modified from the counter cache
 * 3. Reads the rest of the current values (cache misses) from the CF
 * 4. Writes the updated counter values
 * 5. Updates the counter cache
 * 6. Releases the lock(s)
 *
 * See CASSANDRA-4775 and CASSANDRA-6504 for further details.
 *
 * @return the applied resulting Mutation
 */
public Mutation apply() throws WriteTimeoutException
{
    Mutation result = new Mutation(getKeyspaceName(), key());
    Keyspace keyspace = Keyspace.open(getKeyspaceName());

    List<Lock> locks = new ArrayList<>();
    Tracing.trace("Acquiring counter locks");
    try
    {
        grabCounterLocks(keyspace, locks);
        for (PartitionUpdate upd : getPartitionUpdates())
            result.add(processModifications(upd));
        result.apply();
        return result;
    }
    finally
    {
        for (Lock lock : locks)
            lock.unlock();
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public void finish(Set<InetAddress> hintedNodes)
{
    for (int i = 0; i < replayHandlers.size(); i++)
    {
        ReplayWriteResponseHandler<Mutation> handler = replayHandlers.get(i);
        try
        {
            handler.get();
        }
        catch (WriteTimeoutException|WriteFailureException e)
        {
            logger.trace("Failed replaying a batched mutation to a node, will write a hint");
            logger.trace("Failure was : {}", e.getMessage());
            // writing hints for the rest to hints, starting from i
            writeHintsForUndeliveredEndpoints(i, hintedNodes);
            return;
        }
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testSingleCell() throws WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF1);
    cfs.truncateBlocking();
    ColumnDefinition cDef = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("val"));

    // Do the initial update (+1)
    addAndCheck(cfs, 1, 1);

    // Make another increment (+2)
    addAndCheck(cfs, 2, 3);

    // Decrement to 0 (-3)
    addAndCheck(cfs, -3, 0);
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testDuplicateCells() throws WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF1);
    cfs.truncateBlocking();

    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 1L);
    cells.addCounter(cellname(1), 2L);
    cells.addCounter(cellname(1), 3L);
    cells.addCounter(cellname(1), 4L);
    new CounterMutation(new Mutation(KEYSPACE1, bytes(1), cells), ConsistencyLevel.ONE).apply();

    ColumnFamily current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    ByteBuffer context = current.getColumn(cellname(1)).value();
    assertEquals(10L, CounterContext.instance().total(context));
    assertEquals(ClockAndCount.create(1L, 10L), CounterContext.instance().getLocalClockAndCount(context));
    assertEquals(ClockAndCount.create(1L, 10L), cfs.getCachedCounter(bytes(1), cellname(1)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
    CacheService.instance.invalidateCounterCache();

    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addColumn(new BufferCounterUpdateCell(cellname(1), 1L, FBUtilities.timestampMicros()));
    cells.addColumn(new BufferCounterUpdateCell(cellname(2), 2L, FBUtilities.timestampMicros()));
    new CounterMutation(new Mutation(KEYSPACE1, bytes(1), cells), ConsistencyLevel.ONE).apply();
    new CounterMutation(new Mutation(KEYSPACE1, bytes(2), cells), ConsistencyLevel.ONE).apply();

    // flush the counter cache and invalidate
    CacheService.instance.counterCache.submitWrite(Integer.MAX_VALUE).get();
    CacheService.instance.invalidateCounterCache();
    assertEquals(0, CacheService.instance.counterCache.size());

    // load from cache and validate
    CacheService.instance.counterCache.loadSaved(cfs);
    assertEquals(4, CacheService.instance.counterCache.size());
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(1), cellname(2)));
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(2), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(2), cellname(2)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
private void grabCounterLocks(Keyspace keyspace, List<Lock> locks) throws WriteTimeoutException
{
    long startTime = System.nanoTime();

    for (Lock lock : Keyspace.counterLocksFor(getCounterLockKeys()))
    {
        long timeout = TimeUnit.MILLISECONDS.toNanos(getTimeout()) - (System.nanoTime() - startTime);
        try
        {
            if (!lock.tryLock(timeout, TimeUnit.NANOSECONDS))
                throw new WriteTimeoutException(WriteType.COUNTER, consistency(), 0, consistency().blockFor(keyspace));
            locks.add(lock);
        }
        catch (InterruptedException e)
        {
            throw new WriteTimeoutException(WriteType.COUNTER, consistency(), 0, consistency().blockFor(keyspace));
        }
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KS).getColumnFamilyStore(CF);
    CacheService.instance.invalidateCounterCache();

    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addColumn(new CounterUpdateCell(cellname(1), 1L, FBUtilities.timestampMicros()));
    cells.addColumn(new CounterUpdateCell(cellname(2), 2L, FBUtilities.timestampMicros()));
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    new CounterMutation(new Mutation(KS, bytes(2), cells), ConsistencyLevel.ONE).apply();

    // flush the counter cache and invalidate
    CacheService.instance.counterCache.submitWrite(Integer.MAX_VALUE).get();
    CacheService.instance.invalidateCounterCache();
    assertEquals(0, CacheService.instance.counterCache.size());

    // load from cache and validate
    CacheService.instance.counterCache.loadSaved(cfs);
    assertEquals(4, CacheService.instance.counterCache.size());
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(1), cellname(2)));
    assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(2), cellname(1)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(2), cellname(2)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public void await() throws WriteTimeoutException
{
    try
    {
        if (!latch.await(DatabaseDescriptor.getWriteRpcTimeout(), TimeUnit.MILLISECONDS))
            throw new WriteTimeoutException(WriteType.CAS, consistency, getResponseCount(), targets);
    }
    catch (InterruptedException ex)
    {
        throw new AssertionError("This latch shouldn't have been interrupted.");
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
protected void fillCounterCF(ColumnFamilyStore cfs, int rowsPerSSTable) throws WriteTimeoutException
{
    for (int i = 0; i < rowsPerSSTable; i++)
    {
        String key = String.valueOf(i);
        ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, COUNTER_CF);
        Mutation rm = new Mutation(KEYSPACE, ByteBufferUtil.bytes(key), cf);
        rm.addCounter(COUNTER_CF, cellname("Column1"), 100);
        CounterMutation cm = new CounterMutation(rm, ConsistencyLevel.ONE);
        cm.apply();
    }

    cfs.forceBlockingFlush();
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testSingleCell() throws WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KS).getColumnFamilyStore(CF1);
    cfs.truncateBlocking();

    // Do the initial update (+1)
    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 1L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    ColumnFamily current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(1L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));

    // Make another increment (+2)
    cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 2L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(3L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));

    // Decrement to 0 (-3)
    cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), -3L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(0L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));
    assertEquals(ClockAndCount.create(3L, 0L), cfs.getCachedCounter(bytes(1), cellname(1)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testTwoCells() throws WriteTimeoutException
{
    ColumnFamilyStore cfs = Keyspace.open(KS).getColumnFamilyStore(CF1);
    cfs.truncateBlocking();

    // Do the initial update (+1, -1)
    ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 1L);
    cells.addCounter(cellname(2), -1L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    ColumnFamily current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(1L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));
    assertEquals(-1L, CounterContext.instance().total(current.getColumn(cellname(2)).value()));

    // Make another increment (+2, -2)
    cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), 2L);
    cells.addCounter(cellname(2), -2L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(3L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));

    // Decrement to 0 (-3, +3)
    cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    cells.addCounter(cellname(1), -3L);
    cells.addCounter(cellname(2), 3L);
    new CounterMutation(new Mutation(KS, bytes(1), cells), ConsistencyLevel.ONE).apply();
    current = cfs.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    assertEquals(0L, CounterContext.instance().total(current.getColumn(cellname(1)).value()));
    assertEquals(0L, CounterContext.instance().total(current.getColumn(cellname(2)).value()));

    // Check the caches, separately
    assertEquals(ClockAndCount.create(3L, 0L), cfs.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(3L, 0L), cfs.getCachedCounter(bytes(1), cellname(2)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testBatch() throws WriteTimeoutException
{
    ColumnFamilyStore cfs1 = Keyspace.open(KS).getColumnFamilyStore(CF1);
    ColumnFamilyStore cfs2 = Keyspace.open(KS).getColumnFamilyStore(CF2);

    cfs1.truncateBlocking();
    cfs2.truncateBlocking();

    // Do the update (+1, -1), (+2, -2)
    ColumnFamily cells1 = ArrayBackedSortedColumns.factory.create(cfs1.metadata);
    cells1.addCounter(cellname(1), 1L);
    cells1.addCounter(cellname(2), -1L);

    ColumnFamily cells2 = ArrayBackedSortedColumns.factory.create(cfs2.metadata);
    cells2.addCounter(cellname(1), 2L);
    cells2.addCounter(cellname(2), -2L);

    Mutation mutation = new Mutation(KS, bytes(1));
    mutation.add(cells1);
    mutation.add(cells2);

    new CounterMutation(mutation, ConsistencyLevel.ONE).apply();

    // Validate all values
    ColumnFamily current1 = cfs1.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF1, System.currentTimeMillis()));
    ColumnFamily current2 = cfs2.getColumnFamily(QueryFilter.getIdentityFilter(dk(bytes(1)), CF2, System.currentTimeMillis()));

    assertEquals(1L, CounterContext.instance().total(current1.getColumn(cellname(1)).value()));
    assertEquals(-1L, CounterContext.instance().total(current1.getColumn(cellname(2)).value()));
    assertEquals(2L, CounterContext.instance().total(current2.getColumn(cellname(1)).value()));
    assertEquals(-2L, CounterContext.instance().total(current2.getColumn(cellname(2)).value()));

    // Check the caches, separately
    assertEquals(ClockAndCount.create(1L, 1L), cfs1.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(1L, -1L), cfs1.getCachedCounter(bytes(1), cellname(2)));
    assertEquals(ClockAndCount.create(1L, 2L), cfs2.getCachedCounter(bytes(1), cellname(1)));
    assertEquals(ClockAndCount.create(1L, -2L), cfs2.getCachedCounter(bytes(1), cellname(2)));
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public void await() throws WriteTimeoutException
{
    try
    {
        if (!latch.await(DatabaseDescriptor.getWriteRpcTimeout(), TimeUnit.MILLISECONDS))
            throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, getResponseCount(), targets);
    }
    catch (InterruptedException ex)
    {
        throw new AssertionError("This latch shouldn't have been interrupted.");
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
public static void syncWriteToBatchlog(WriteResponseHandler<?> handler, Batch batch, Collection<InetAddress> endpoints)
throws WriteTimeoutException, WriteFailureException
{
    for (InetAddress target : endpoints)
    {
        logger.trace("Sending legacy batchlog store request {} to {} for {} mutations", batch.id, target, batch.size());

        int targetVersion = MessagingService.instance().getVersion(target);
        MessagingService.instance().sendRR(getStoreMutation(batch, targetVersion).createMessage(MessagingService.Verb.MUTATION),
                                           target,
                                           handler,
                                           false);
    }
}
 

import org.apache.cassandra.exceptions.WriteTimeoutException; //导入依赖的package包/类
@Test
public void testScrubCorruptedCounterRowNoEarlyOpen() throws IOException, WriteTimeoutException
{
    boolean oldDisabledVal = SSTableRewriter.disableEarlyOpeningForTests;
    try
    {
        SSTableRewriter.disableEarlyOpeningForTests = true;
        testScrubCorruptedCounterRow();
    }
    finally
    {
        SSTableRewriter.disableEarlyOpeningForTests = oldDisabledVal;
    }
}