import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private void repairKeyspace(String keyspace, List<String> columnFamilies) throws Exception {
    LOGGER.info("Starting repair : keySpace = {}, columnFamilies = {}",
        keyspace, columnFamilies);

    Map<String, String> options = new HashMap<>();
    options.put(RepairOption.PRIMARY_RANGE_KEY, "true");
    options.put(RepairOption.COLUMNFAMILIES_KEY,
        String.join(",", columnFamilies));
    options.put(RepairOption.PARALLELISM_KEY,
        RepairParallelism.SEQUENTIAL.getName());
    options.put(RepairOption.INCREMENTAL_KEY, "true");

    String result = daemon.repair(keyspace, options);

    LOGGER.info("Repair output = {}", result);
    LOGGER.info("Completed repair : keySpace = {}, columnFamilies = {}",
        keyspace, columnFamilies);

    sendStatus(driver, Protos.TaskState.TASK_RUNNING,
        String.format(
            "Completed repair : keySpace = %s, columnFamilies = %s",
            keyspace, columnFamilies));
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
@Override
public RepairRun map(int index, ResultSet rs, StatementContext ctx) throws SQLException {
  RepairRun.RunState runState = RepairRun.RunState.valueOf(rs.getString("state"));
  RepairParallelism repairParallelism = RepairParallelism.fromName(
      rs.getString("repair_parallelism").toLowerCase().replace("datacenter_aware", "dc_parallel"));

  RepairRun.Builder repairRunBuilder = new RepairRun.Builder(
      rs.getString("cluster_name"),
      UuidUtil.fromSequenceId(rs.getLong("repair_unit_id")),
      getDateTimeOrNull(rs, "creation_time"),
      rs.getFloat("intensity"),
      rs.getInt("segment_count"),
      repairParallelism);

  return repairRunBuilder
      .runState(runState)
      .owner(rs.getString("owner"))
      .cause(rs.getString("cause"))
      .startTime(getDateTimeOrNull(rs, "start_time"))
      .endTime(getDateTimeOrNull(rs, "end_time"))
      .pauseTime(getDateTimeOrNull(rs, "pause_time"))
      .lastEvent(rs.getString("last_event"))
      .build(UuidUtil.fromSequenceId(rs.getLong("id")));
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private RepairSchedule createRepairScheduleFromRow(Row repairScheduleRow) {
  return new RepairSchedule.Builder(
          repairScheduleRow.getUUID("repair_unit_id"),
          RepairSchedule.State.valueOf(repairScheduleRow.getString("state")),
          repairScheduleRow.getInt("days_between"),
          new DateTime(repairScheduleRow.getTimestamp("next_activation")),
          ImmutableList.copyOf(repairScheduleRow.getSet("run_history", UUID.class)),
          repairScheduleRow.getInt("segment_count"),
          RepairParallelism.fromName(repairScheduleRow.getString("repair_parallelism")),
          repairScheduleRow.getDouble("intensity"),
          new DateTime(repairScheduleRow.getTimestamp("creation_time")),
          repairScheduleRow.getInt("segment_count_per_node"))
      .owner(repairScheduleRow.getString("owner"))
      .pauseTime(new DateTime(repairScheduleRow.getTimestamp("pause_time")))
      .build(repairScheduleRow.getUUID("id"));
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private RepairRun buildRepairRunFromRow(Row repairRunResult, UUID id) {
  LOG.trace("buildRepairRunFromRow {} / {}", id, repairRunResult);
  return new RepairRun.Builder(
      repairRunResult.getString("cluster_name"),
      repairRunResult.getUUID("repair_unit_id"),
      new DateTime(repairRunResult.getTimestamp("creation_time")),
      repairRunResult.getDouble("intensity"),
      repairRunResult.getInt("segment_count"),
      RepairParallelism.fromName(repairRunResult.getString("repair_parallelism")))
      .cause(repairRunResult.getString("cause"))
      .owner(repairRunResult.getString("owner"))
      .endTime(new DateTime(repairRunResult.getTimestamp("end_time")))
      .lastEvent(repairRunResult.getString("last_event"))
      .pauseTime(new DateTime(repairRunResult.getTimestamp("pause_time")))
      .runState(RunState.valueOf(repairRunResult.getString("state")))
      .startTime(new DateTime(repairRunResult.getTimestamp("start_time")))
      .build(id);
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public Builder(
    String clusterName,
    UUID repairUnitId,
    DateTime creationTime,
    double intensity,
    int segmentCount,
    RepairParallelism repairParallelism) {

  this.clusterName = clusterName;
  this.repairUnitId = repairUnitId;
  this.runState = RunState.NOT_STARTED;
  this.creationTime = creationTime;
  this.intensity = intensity;
  this.segmentCount = segmentCount;
  this.repairParallelism = repairParallelism;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public Builder(
    UUID repairUnitId,
    State state,
    int daysBetween,
    DateTime nextActivation,
    ImmutableList<UUID> runHistory,
    int segmentCount,
    RepairParallelism repairParallelism,
    double intensity,
    DateTime creationTime,
    int segmentCountPerNode) {
  this.repairUnitId = repairUnitId;
  this.state = state;
  this.daysBetween = daysBetween;
  this.nextActivation = nextActivation;
  this.runHistory = runHistory;
  this.segmentCount = segmentCount;
  this.repairParallelism = repairParallelism;
  this.intensity = intensity;
  this.creationTime = creationTime;
  this.segmentCountPerNode = segmentCountPerNode;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
@Before
public void setUp() {
  //create a valid config
  DataSourceFactory dataSourceFactory = new DataSourceFactory();
  dataSourceFactory.setDriverClass("org.postgresql.Driver");
  dataSourceFactory.setUrl("jdbc:postgresql://db.example.com/db-prod");
  dataSourceFactory.setUser("user");
  CassandraFactory cassandraFactory = new CassandraFactory();
  cassandraFactory.setContactPoints(new String[]{"127.0.0.1"});
  config.setCassandraFactory(cassandraFactory);
  config.setPostgresDataSourceFactory(dataSourceFactory);
  config.setHangingRepairTimeoutMins(1);
  config.setRepairParallelism(RepairParallelism.DATACENTER_AWARE);
  config.setRepairRunThreadCount(1);
  config.setSegmentCount(1);
  config.setScheduleDaysBetween(7);
  config.setStorageType("foo");
  config.setIncrementalRepair(false);
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public int forceRepairAsync(String keyspace, int parallelismDegree, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies)
{
    if (parallelismDegree < 0 || parallelismDegree > RepairParallelism.values().length - 1)
    {
        throw new IllegalArgumentException("Invalid parallelism degree specified: " + parallelismDegree);
    }
    Collection<Range<Token>> ranges;
    if (primaryRange)
    {
        // when repairing only primary range, neither dataCenters nor hosts can be set
        if (dataCenters == null && hosts == null)
            ranges = getPrimaryRanges(keyspace);
        // except dataCenters only contain local DC (i.e. -local)
        else if (dataCenters != null && dataCenters.size() == 1 && dataCenters.contains(DatabaseDescriptor.getLocalDataCenter()))
            ranges = getPrimaryRangesWithinDC(keyspace);
        else
            throw new IllegalArgumentException("You need to run primary range repair on all nodes in the cluster.");
    }
    else
    {
         ranges = getLocalRanges(keyspace);
    }

    return forceRepairAsync(keyspace, RepairParallelism.values()[parallelismDegree], dataCenters, hosts, ranges, fullRepair, columnFamilies);
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean isSubrangeRepair)
{
    if (FBUtilities.isWindows() &&
        (DatabaseDescriptor.getDiskAccessMode() != Config.DiskAccessMode.standard || DatabaseDescriptor.getIndexAccessMode() != Config.DiskAccessMode.standard) &&
        parallelism == RepairParallelism.SEQUENTIAL)
    {
        logger.warn("Sequential repair disabled when memory-mapped I/O is configured on Windows. Reverting to parallel.");
        this.parallelism = RepairParallelism.PARALLEL;
    }
    else
        this.parallelism = parallelism;

    this.primaryRange = primaryRange;
    this.incremental = incremental;
    this.trace = trace;
    this.jobThreads = jobThreads;
    this.ranges.addAll(ranges);
    this.isSubrangeRepair = isSubrangeRepair;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
Map<String,String> getOptions() {
    Map<String, String> result = new HashMap<>();
    result.put(RepairOption.PARALLELISM_KEY, String.valueOf(RepairParallelism.SEQUENTIAL));
    result.put(RepairOption.HOSTS_KEY, endpoint);
    if(!table.equals("") && !table.equals("*")) {
        result.put(RepairOption.COLUMNFAMILIES_KEY, table);
    }
    result.put(RepairOption.RANGES_KEY, String.format("%s:%s", start, end));
    return result;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private static void checkRepairParallelismString(String repairParallelism) throws ReaperException {
  try {
    RepairParallelism.valueOf(repairParallelism.toUpperCase());
  } catch (IllegalArgumentException ex) {
    throw new ReaperException(
        "invalid repair parallelism given \""
        + repairParallelism
        + "\", must be one of: "
        + Arrays.toString(RepairParallelism.values()),
        ex);
  }
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public RepairScheduleStatus(
    UUID id,
    String owner,
    String clusterName,
    String keyspaceName,
    Collection<String> columnFamilies,
    RepairSchedule.State state,
    DateTime creationTime,
    DateTime nextActivation,
    DateTime pauseTime,
    double intensity,
    boolean incrementalRepair,
    int segmentCount,
    RepairParallelism repairParallelism,
    int daysBetween,
    Collection<String> nodes,
    Collection<String> datacenters,
    Collection<String> blacklistedTables,
    int segmentCountPerNode) {

  this.id = id;
  this.owner = owner;
  this.clusterName = clusterName;
  this.keyspaceName = keyspaceName;
  this.columnFamilies = columnFamilies;
  this.state = state;
  this.creationTime = creationTime;
  this.nextActivation = nextActivation;
  this.pauseTime = pauseTime;
  this.intensity = RepairRunStatus.roundDoubleNicely(intensity);
  this.incrementalRepair = incrementalRepair;
  this.segmentCount = segmentCount;
  this.repairParallelism = repairParallelism;
  this.daysBetween = daysBetween;
  this.nodes = nodes;
  this.datacenters = datacenters;
  this.blacklistedTables = blacklistedTables;
  this.segmentCountPerNode = segmentCountPerNode;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
/**
 * Triggers a repair of range (beginToken, endToken] for given keyspace and column family. The repair is triggered by
 * {@link org.apache.cassandra.service.StorageServiceMBean#forceRepairRangeAsync} For time being, we don't allow local
 * nor snapshot repairs.
 *
 * @return Repair command number, or 0 if nothing to repair
 */
int triggerRepair(
    BigInteger beginToken,
    BigInteger endToken,
    String keyspace,
    RepairParallelism repairParallelism,
    Collection<String> columnFamilies,
    boolean fullRepair,
    Collection<String> datacenters) throws ReaperException;
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private int triggerRepairPost2dot2(
    boolean fullRepair,
    RepairParallelism repairParallelism,
    String keyspace,
    Collection<String> columnFamilies,
    BigInteger beginToken,
    BigInteger endToken,
    String cassandraVersion,
    Collection<String> datacenters) {

  Map<String, String> options = new HashMap<>();

  options.put(RepairOption.PARALLELISM_KEY, repairParallelism.getName());
  // options.put(RepairOption.PRIMARY_RANGE_KEY, Boolean.toString(primaryRange));
  options.put(RepairOption.INCREMENTAL_KEY, Boolean.toString(!fullRepair));
  options.put(RepairOption.JOB_THREADS_KEY, Integer.toString(1));
  options.put(RepairOption.TRACE_KEY, Boolean.toString(Boolean.FALSE));
  options.put(RepairOption.COLUMNFAMILIES_KEY, StringUtils.join(columnFamilies, ","));
  // options.put(RepairOption.PULL_REPAIR_KEY, Boolean.FALSE);
  if (fullRepair) {
    options.put(RepairOption.RANGES_KEY, beginToken.toString() + ":" + endToken.toString());
  }

  options.put(RepairOption.DATACENTERS_KEY, StringUtils.join(datacenters, ","));
  // options.put(RepairOption.HOSTS_KEY, StringUtils.join(specificHosts, ","));

  return ((StorageServiceMBean) ssProxy).repairAsync(keyspace, options);
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
private int triggerRepairPre2dot1(
    RepairParallelism repairParallelism,
    String keyspace,
    Collection<String> columnFamilies,
    BigInteger beginToken,
    BigInteger endToken,
    Collection<String> datacenters) {

  // Cassandra 1.2 and 2.0 compatibility
  if (repairParallelism.equals(RepairParallelism.DATACENTER_AWARE)) {
    return ((StorageServiceMBean20) ssProxy)
        .forceRepairRangeAsync(
            beginToken.toString(),
            endToken.toString(),
            keyspace,
            repairParallelism.ordinal(),
            datacenters,
            null,
            columnFamilies.toArray(new String[columnFamilies.size()]));
  }
  boolean snapshotRepair = repairParallelism.equals(RepairParallelism.SEQUENTIAL);
  return ((StorageServiceMBean20) ssProxy)
      .forceRepairRangeAsync(
          beginToken.toString(),
          endToken.toString(),
          keyspace,
          snapshotRepair,
          false,
          columnFamilies.toArray(new String[columnFamilies.size()]));
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
/**
 * Instantiates a RepairRun and stores it in the storage backend.
 *
 * @return the new, just stored RepairRun instance
 * @throws ReaperException when fails to store the RepairRun.
 */
private static RepairRun.Builder createNewRepairRun(
    Cluster cluster,
    RepairUnit repairUnit,
    Optional<String> cause,
    String owner,
    int segments,
    RepairParallelism repairParallelism,
    Double intensity) throws ReaperException {

  return new RepairRun.Builder(
      cluster.getName(), repairUnit.getId(), DateTime.now(), intensity, segments, repairParallelism)
      .cause(cause.or("no cause specified"))
      .owner(owner);
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
SegmentRunner(
    AppContext context,
    UUID segmentId,
    Collection<String> potentialCoordinators,
    long timeoutMillis,
    double intensity,
    RepairParallelism validationParallelism,
    String clusterName,
    RepairUnit repairUnit,
    RepairRunner repairRunner)
    throws ReaperException {

  if (SEGMENT_RUNNERS.containsKey(segmentId)) {
    LOG.error("SegmentRunner already exists for segment with ID: {}", segmentId);
    throw new ReaperException("SegmentRunner already exists for segment with ID: " + segmentId);
  }
  this.context = context;
  this.segmentId = segmentId;
  this.potentialCoordinators = potentialCoordinators;
  this.timeoutMillis = timeoutMillis;
  this.intensity = intensity;
  this.validationParallelism = validationParallelism;
  this.clusterName = clusterName;
  this.repairUnit = repairUnit;
  this.repairRunner = repairRunner;
  this.segmentFailed = new AtomicBoolean(false);
  this.leaderElectionId = repairUnit.getIncrementalRepair() ? repairRunner.getRepairRunId() : segmentId;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
public RepairParameters(
    RingRange tokenRange,
    String keyspaceName,
    Set<String> columnFamilies,
    RepairParallelism repairParallelism) {

  this.tokenRange = tokenRange;
  this.keyspaceName = keyspaceName;
  this.columnFamilies = columnFamilies;
  this.repairParallelism = repairParallelism;
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
@Override
public RepairScheduleStatus map(int index, ResultSet rs, StatementContext ctx) throws SQLException {

  return new RepairScheduleStatus(
      UuidUtil.fromSequenceId(rs.getLong("id")),
      rs.getString("owner"),
      rs.getString("cluster_name"),
      rs.getString("keyspace_name"),
      ImmutableSet.copyOf(getStringArray(rs.getArray("column_families").getArray())),
      RepairSchedule.State.valueOf(rs.getString("state")),
      RepairRunMapper.getDateTimeOrNull(rs, "creation_time"),
      RepairRunMapper.getDateTimeOrNull(rs, "next_activation"),
      RepairRunMapper.getDateTimeOrNull(rs, "pause_time"),
      rs.getDouble("intensity"),
      rs.getBoolean("incremental_repair"),
      rs.getInt("segment_count"),
      RepairParallelism.fromName(
          rs.getString("repair_parallelism")
              .toLowerCase()
              .replace("datacenter_aware", "dc_parallel")),
      rs.getInt("days_between"),
      ImmutableSet.copyOf(
          rs.getArray("nodes") == null
              ? new String[] {}
              : getStringArray(rs.getArray("nodes").getArray())),
      ImmutableSet.copyOf(
          rs.getArray("datacenters") == null
              ? new String[] {}
              : getStringArray(rs.getArray("datacenters").getArray())),
      ImmutableSet.copyOf(
          rs.getArray("blacklisted_tables") == null
              ? new String[] {}
              : getStringArray(rs.getArray("blacklisted_tables").getArray())),
      rs.getInt("segment_count_per_node"));
}
 

import org.apache.cassandra.repair.RepairParallelism; //导入依赖的package包/类
@Override
public RepairParameters map(int index, ResultSet rs, StatementContext ctx) throws SQLException {

  RingRange range
      = new RingRange(rs.getBigDecimal("start_token").toBigInteger(), rs.getBigDecimal("end_token").toBigInteger());

  Object columnFamiliesObj = rs.getArray("column_families").getArray();
  String[] columnFamilies;
  if (columnFamiliesObj instanceof String[]) {
    columnFamilies = (String[]) columnFamiliesObj;
  } else {
    Object[] objArray = (Object[]) columnFamiliesObj;
    columnFamilies = Arrays.copyOf(objArray, objArray.length, String[].class);
  }

  String repairParallelismStr = rs.getString("repair_parallelism");
  if (repairParallelismStr != null) {
    repairParallelismStr = repairParallelismStr.toUpperCase();
  }
  RepairParallelism repairParallelism = RepairParallelism.fromName(repairParallelismStr);

  return new RepairParameters(
      range,
      rs.getString("keyspace_name"),
      Sets.newHashSet(columnFamilies),
      repairParallelism);
}