import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?";
    UntypedResultSet results = executeInternal(String.format(req, PAXOS_CF), key, metadata.cfId);
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata));
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
public static PaxosState loadPaxosState(DecoratedKey key, CFMetaData metadata, int nowInSec)
{
    String req = "SELECT * FROM system.%s WHERE row_key = ? AND cf_id = ?";
    UntypedResultSet results = QueryProcessor.executeInternalWithNow(nowInSec, String.format(req, PAXOS), key.getKey(), metadata.cfId);
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(row.getUUID("in_progress_ballot"), new PartitionUpdate(metadata, key, metadata.partitionColumns(), 1))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    int proposalVersion = row.has("proposal_version") ? row.getInt("proposal_version") : MessagingService.VERSION_21;
    Commit accepted = row.has("proposal")
                    ? new Commit(row.getUUID("proposal_ballot"), PartitionUpdate.fromBytes(row.getBytes("proposal"), proposalVersion, key))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    int mostRecentVersion = row.has("most_recent_commit_version") ? row.getInt("most_recent_commit_version") : MessagingService.VERSION_21;
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(row.getUUID("most_recent_commit_at"), PartitionUpdate.fromBytes(row.getBytes("most_recent_commit"), mostRecentVersion, key))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
public static PaxosState loadPaxosState(ByteBuffer key, CFMetaData metadata)
{
    String req = "SELECT * FROM system.%s WHERE row_key = 0x%s AND cf_id = %s";
    UntypedResultSet results = processInternal(String.format(req, PAXOS_CF, ByteBufferUtil.bytesToHex(key), metadata.cfId));
    if (results.isEmpty())
        return new PaxosState(key, metadata);
    UntypedResultSet.Row row = results.one();
    Commit promised = row.has("in_progress_ballot")
                    ? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
                    : Commit.emptyCommit(key, metadata);
    // either we have both a recently accepted ballot and update or we have neither
    Commit accepted = row.has("proposal")
                    ? new Commit(key, row.getUUID("proposal_ballot"), ColumnFamily.fromBytes(row.getBytes("proposal")))
                    : Commit.emptyCommit(key, metadata);
    // either most_recent_commit and most_recent_commit_at will both be set, or neither
    Commit mostRecent = row.has("most_recent_commit")
                      ? new Commit(key, row.getUUID("most_recent_commit_at"), ColumnFamily.fromBytes(row.getBytes("most_recent_commit")))
                      : Commit.emptyCommit(key, metadata);
    return new PaxosState(promised, accepted, mostRecent);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
@Test
public void testCommittingAfterTruncation() throws Exception
{
    ColumnFamilyStore cfs = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard1");
    DecoratedKey key = Util.dk("key" + System.nanoTime());
    CellName name = Util.cellname("col");
    ByteBuffer value = ByteBufferUtil.bytes(0);
    ColumnFamily update = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    update.addColumn(name, value, FBUtilities.timestampMicros());

    // CFS should be empty initially
    assertNoDataPresent(cfs, key);

    // Commit the proposal & verify the data is present
    Commit beforeTruncate = newProposal(0, key.getKey(), update);
    PaxosState.commit(beforeTruncate);
    assertDataPresent(cfs, key, name, value);

    // Truncate then attempt to commit again, mutation should
    // be ignored as the proposal predates the truncation
    cfs.truncateBlocking();
    PaxosState.commit(beforeTruncate);
    assertNoDataPresent(cfs, key);

    // Now try again with a ballot created after the truncation
    long timestamp = SystemKeyspace.getTruncatedAt(update.metadata().cfId) + 1;
    Commit afterTruncate = newProposal(timestamp, key.getKey(), update);
    PaxosState.commit(afterTruncate);
    assertDataPresent(cfs, key, name, value);
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
/**
 * Commit a PAXOS task locally, and if the task times out rather then submitting a real hint
 * submit a fake one that executes immediately on the mutation stage, but generates the necessary backpressure
 * signal for hints
 */
private static void commitPaxosLocal(final MessageOut<Commit> message, final AbstractWriteResponseHandler<?> responseHandler)
{
    StageManager.getStage(MessagingService.verbStages.get(MessagingService.Verb.PAXOS_COMMIT)).maybeExecuteImmediately(new LocalMutationRunnable()
    {
        public void runMayThrow()
        {
            try
            {
                PaxosState.commit(message.payload);
                if (responseHandler != null)
                    responseHandler.response(null);
            }
            catch (Exception ex)
            {
                if (!(ex instanceof WriteTimeoutException))
                    logger.error("Failed to apply paxos commit locally : {}", ex);
                responseHandler.onFailure(FBUtilities.getBroadcastAddress());
            }
        }

        @Override
        protected Verb verb()
        {
            return MessagingService.Verb.PAXOS_COMMIT;
        }
    });
}
 

import org.apache.cassandra.service.paxos.PaxosState; //导入依赖的package包/类
@Test
public void testCommittingAfterTruncation() throws Exception
{
    ColumnFamilyStore cfs = Keyspace.open("PaxosStateTestKeyspace1").getColumnFamilyStore("Standard1");
    String key = "key" + System.nanoTime();
    ByteBuffer value = ByteBufferUtil.bytes(0);
    RowUpdateBuilder builder = new RowUpdateBuilder(cfs.metadata, FBUtilities.timestampMicros(), key);
    builder.clustering("a").add("val", value);
    PartitionUpdate update = Iterables.getOnlyElement(builder.build().getPartitionUpdates());

    // CFS should be empty initially
    assertNoDataPresent(cfs, Util.dk(key));

    // Commit the proposal & verify the data is present
    Commit beforeTruncate = newProposal(0, update);
    PaxosState.commit(beforeTruncate);
    assertDataPresent(cfs, Util.dk(key), "val", value);

    // Truncate then attempt to commit again, mutation should
    // be ignored as the proposal predates the truncation
    cfs.truncateBlocking();
    PaxosState.commit(beforeTruncate);
    assertNoDataPresent(cfs, Util.dk(key));

    // Now try again with a ballot created after the truncation
    long timestamp = SystemKeyspace.getTruncatedAt(update.metadata().cfId) + 1;
    Commit afterTruncate = newProposal(timestamp, update);
    PaxosState.commit(afterTruncate);
    assertDataPresent(cfs, Util.dk(key), "val", value);
}