import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public DatacenterSyncWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                          Collection<InetAddress> pendingEndpoints,
                                          ConsistencyLevel consistencyLevel,
                                          Keyspace keyspace,
                                          Runnable callback,
                                          WriteType writeType)
{
    // Response is been managed by the map so make it 1 for the superclass.
    super(keyspace, naturalEndpoints, pendingEndpoints, consistencyLevel, callback, writeType);
    assert consistencyLevel == ConsistencyLevel.EACH_QUORUM;

    strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();

    for (String dc : strategy.getDatacenters())
    {
        int rf = strategy.getReplicationFactor(dc);
        responses.put(dc, new AtomicInteger((rf / 2) + 1));
    }

    // During bootstrap, we have to include the pending endpoints or we may fail the consistency level
    // guarantees (see #833)
    for (InetAddress pending : pendingEndpoints)
    {
        responses.get(snitch.getDatacenter(pending)).incrementAndGet();
    }
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
/**
 * @param callback A callback to be called when the write is successful.
 */
protected AbstractWriteResponseHandler(Keyspace keyspace,
                                       Collection<InetAddress> naturalEndpoints,
                                       Collection<InetAddress> pendingEndpoints,
                                       ConsistencyLevel consistencyLevel,
                                       Runnable callback,
                                       WriteType writeType)
{
    this.keyspace = keyspace;
    this.pendingEndpoints = pendingEndpoints;
    this.start = System.nanoTime();
    this.consistencyLevel = consistencyLevel;
    this.naturalEndpoints = naturalEndpoints;
    this.callback = callback;
    this.writeType = writeType;
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的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.db.WriteType; //导入依赖的package包/类
/**
 * @param pendingEndpoints
 * @param callback A callback to be called when the write is successful.
 */
protected AbstractWriteResponseHandler(Keyspace keyspace,
                                       Collection<InetAddress> naturalEndpoints,
                                       Collection<InetAddress> pendingEndpoints,
                                       ConsistencyLevel consistencyLevel,
                                       Runnable callback,
                                       WriteType writeType)
{
    this.keyspace = keyspace;
    this.pendingEndpoints = pendingEndpoints;
    this.start = System.nanoTime();
    this.consistencyLevel = consistencyLevel;
    this.naturalEndpoints = naturalEndpoints;
    this.callback = callback;
    this.writeType = writeType;
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public <T> AbstractWriteResponseHandler<T> getWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                                            Collection<InetAddress> pendingEndpoints,
                                                            ConsistencyLevel consistency_level,
                                                            Runnable callback,
                                                            WriteType writeType)
{
    if (consistency_level.isDatacenterLocal())
    {
        // block for in this context will be localnodes block.
        return new DatacenterWriteResponseHandler<T>(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    else if (consistency_level == ConsistencyLevel.EACH_QUORUM && (this instanceof NetworkTopologyStrategy))
    {
        return new DatacenterSyncWriteResponseHandler<T>(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    return new WriteResponseHandler<T>(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public DatacenterSyncWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                          Collection<InetAddress> pendingEndpoints,
                                          ConsistencyLevel consistencyLevel,
                                          Keyspace keyspace,
                                          Runnable callback,
                                          WriteType writeType)
{
    // Response is been managed by the map so make it 1 for the superclass.
    super(keyspace, naturalEndpoints, pendingEndpoints, consistencyLevel, callback, writeType);
    assert consistencyLevel == ConsistencyLevel.EACH_QUORUM;

    NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();

    for (String dc : strategy.getDatacenters())
    {
        int rf = strategy.getReplicationFactor(dc);
        responses.put(dc, new AtomicInteger((rf / 2) + 1));
    }

    // During bootstrap, we have to include the pending endpoints or we may fail the consistency level
    // guarantees (see #833)
    for (InetAddress pending : pendingEndpoints)
    {
        responses.get(snitch.getDatacenter(pending)).incrementAndGet();
    }
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public DatacenterSyncWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                          Collection<InetAddress> pendingEndpoints,
                                          ConsistencyLevel consistencyLevel,
                                          Table table,
                                          Runnable callback,
                                          WriteType writeType)
{
    // Response is been managed by the map so make it 1 for the superclass.
    super(table, naturalEndpoints, pendingEndpoints, consistencyLevel, callback, writeType);
    assert consistencyLevel == ConsistencyLevel.EACH_QUORUM;

    strategy = (NetworkTopologyStrategy) table.getReplicationStrategy();

    for (String dc : strategy.getDatacenters())
    {
        int rf = strategy.getReplicationFactor(dc);
        responses.put(dc, new AtomicInteger((rf / 2) + 1));
    }

    // During bootstrap, we have to include the pending endpoints or we may fail the consistency level
    // guarantees (see #833)
    for (InetAddress pending : pendingEndpoints)
    {
        responses.get(snitch.getDatacenter(pending)).incrementAndGet();
    }
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
/**
 * @param pendingEndpoints
 * @param callback A callback to be called when the write is successful.
 */
protected AbstractWriteResponseHandler(Table table,
                                       Collection<InetAddress> naturalEndpoints,
                                       Collection<InetAddress> pendingEndpoints,
                                       ConsistencyLevel consistencyLevel,
                                       Runnable callback,
                                       WriteType writeType)
{
    this.table = table;
    this.pendingEndpoints = pendingEndpoints;
    this.startTime = System.currentTimeMillis();
    this.consistencyLevel = consistencyLevel;
    this.naturalEndpoints = naturalEndpoints;
    this.callback = callback;
    this.writeType = writeType;
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints, ConsistencyLevel consistency_level, Runnable callback, WriteType writeType)
{
    if (consistency_level.isDatacenterLocal())
    {
        // block for in this context will be localnodes block.
        return new DatacenterWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    else if (consistency_level == ConsistencyLevel.EACH_QUORUM && (this instanceof NetworkTopologyStrategy))
    {
        return new DatacenterSyncWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    return new WriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public DatacenterWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                      Collection<InetAddress> pendingEndpoints,
                                      ConsistencyLevel consistencyLevel,
                                      Keyspace keyspace,
                                      Runnable callback,
                                      WriteType writeType)
{
    super(naturalEndpoints, pendingEndpoints, consistencyLevel, keyspace, callback, writeType);
    assert consistencyLevel.isDatacenterLocal();
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的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.db.WriteType; //导入依赖的package包/类
public WriteResponseHandler(Collection<InetAddress> writeEndpoints,
                            Collection<InetAddress> pendingEndpoints,
                            ConsistencyLevel consistencyLevel,
                            Keyspace keyspace,
                            Runnable callback,
                            WriteType writeType)
{
    super(keyspace, writeEndpoints, pendingEndpoints, consistencyLevel, callback, writeType);
    responses = totalBlockFor();
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public void get() throws WriteTimeoutException
{
    long requestTimeout = writeType == WriteType.COUNTER
                        ? DatabaseDescriptor.getCounterWriteRpcTimeout()
                        : DatabaseDescriptor.getWriteRpcTimeout();

    long timeout = TimeUnit.MILLISECONDS.toNanos(requestTimeout) - (System.nanoTime() - start);

    boolean success;
    try
    {
        success = condition.await(timeout, TimeUnit.NANOSECONDS);
    }
    catch (InterruptedException ex)
    {
        throw new AssertionError(ex);
    }

    if (!success)
    {
        int acks = ackCount();
        int blockedFor = totalBlockFor();
        // It's pretty unlikely, but we can race between exiting await above and here, so
        // that we could now have enough acks. In that case, we "lie" on the acks count to
        // avoid sending confusing info to the user (see CASSANDRA-6491).
        if (acks >= blockedFor)
            acks = blockedFor - 1;
        throw new WriteTimeoutException(writeType, consistencyLevel, acks, blockedFor);
    }
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints, ConsistencyLevel consistency_level, Runnable callback, WriteType writeType)
{
    if (consistency_level == ConsistencyLevel.LOCAL_QUORUM)
    {
        // block for in this context will be localnodes block.
        return new DatacenterWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    else if (consistency_level == ConsistencyLevel.EACH_QUORUM)
    {
        return new DatacenterSyncWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
    }
    return new WriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的package包/类
public DatacenterWriteResponseHandler(Collection<InetAddress> naturalEndpoints,
                                      Collection<InetAddress> pendingEndpoints,
                                      ConsistencyLevel consistencyLevel,
                                      Keyspace keyspace,
                                      Runnable callback,
                                      WriteType writeType)
{
    super(naturalEndpoints, pendingEndpoints, consistencyLevel, keyspace, callback, writeType);
    assert consistencyLevel == ConsistencyLevel.LOCAL_QUORUM;
}
 

import org.apache.cassandra.db.WriteType; //导入依赖的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.db.WriteType; //导入依赖的package包/类
public WriteResponseHandler(Collection<InetAddress> writeEndpoints,
                            Collection<InetAddress> pendingEndpoints,
                            ConsistencyLevel consistencyLevel,
                            Keyspace keyspace,
                            Runnable callback,
                            WriteType writeType)
{
    super(keyspace, writeEndpoints, pendingEndpoints, consistencyLevel, callback, writeType);
    responses = new AtomicInteger(totalBlockFor());
}