Subrange incremental repair (#1509)

* Add support for subrange incremental repairs

pom.xml

@@ -309,7 +309,7 @@
                   <dependency>
                     <groupId>com.puppycrawl.tools</groupId>
                     <artifactId>checkstyle</artifactId>
-                    <version>8.19</version>
+                    <version>10.17.0</version>
                   </dependency>
                 </dependencies>
             </plugin>

src/docs/content/docs/api.md

@@ -196,7 +196,8 @@ Returns OK if all goes well NOT_MODIFIED if new state is the same as the old one
 	    * *segmentCountPerNode*: Defines the amount of segments per node to create for scheduled repair runs. (Optional)
 	    * *repairParallelism*: Defines the used repair parallelism for scheduled repair runs. (Optional)
 	    * *intensity*: Defines the repair intensity for scheduled repair runs. (Optional)
-	    * *incrementalRepair*: Defines if incremental repair should be done. [true/false] (Optional)
+	    * *incrementalRepair*: Defines if incremental repair should be done on all tokens of each node at once. [true/false] (Optional)
+      * *subrangeIncrementalRepair*: Defines if incremental repair should be done in subrange mode, against discrete token ranges. [true/false] (Optional)
 	    * *scheduleDaysBetween*: Defines the amount of days to wait between scheduling new repairs.
 	                             For example, use value 7 for weekly schedule, and 0 for continuous.
 	    * *scheduleTriggerTime*: Defines the time for first scheduled trigger for the run.

src/docs/content/docs/configuration/docker_vars.md

@@ -27,6 +27,7 @@ The Docker environment variables listed in this section map directly to Reaper s
 <code class="codeLarge">REAPER_ENABLE_DYNAMIC_SEED_LIST</code> | [enableDynamicSeedList]({{< relref "reaper_specific.md#enabledynamicseedlist" >}})                                   | true
 <code class="codeLarge">REAPER_HANGING_REPAIR_TIMEOUT_MINS</code> | [hangingRepairTimeoutMins]({{< relref "reaper_specific.md#hangingrepairtimeoutmins" >}})                             | 30
 <code class="codeLarge">REAPER_INCREMENTAL_REPAIR</code> | [incrementalRepair]({{< relref "reaper_specific.md#incrementalrepair" >}})                                           | false
+<code class="codeLarge">REAPER_SUBRANGE_INCREMENTAL</code> | [subrangeIncrementalRepair]({{< relref "reaper_specific.md#subrangeincremental" >}})                                           | false
 <code class="codeLarge">REAPER_JMX_AUTH_PASSWORD</code> | [password]({{< relref "reaper_specific.md#password" >}})                                                             |
 <code class="codeLarge">REAPER_JMX_AUTH_USERNAME</code> | [username]({{< relref "reaper_specific.md#username" >}})                                                             |
 <code class="codeLarge">REAPER_JMX_CREDENTIALS</code> | [jmxCredentials]({{< relref "reaper_specific.md#jmxcredentials" >}})                                                 |

src/docs/content/docs/configuration/reaper_specific.md

@@ -154,6 +154,21 @@ Sets the default repair type unless specifically defined for each run. Note that
 
 <br/>
 
+### `subrangeIncrementalRepair`
+
+Type: *Boolean*
+
+Default: *false*
+
+Sets the default repair type unless specifically defined for each run. Note that this is only supported with the PARALLEL repairParallelism setting. For more details in incremental repair, please refer to the following article.http://www.datastax.com/dev/blog/more-efficient-repairs.
+This mode will split the repair jobs into sets of token ranges using the incremental mode.
+This will prevail over the `incrementalRepair` setting.
+
+
+*Note*: Subrange incremental repair is only available since Cassandra 4.0.
+
+<br/>
+
 ### `blacklistTwcsTables`
 
 Type: *Boolean*

src/packaging/bin/spreaper

@@ -233,6 +233,9 @@ def _arguments_for_repair_and_schedule(parser):
     parser.add_argument("--incremental", default="false",
                         help=("Incremental repair (true or false), "
                               "or use the configured default if not given (false)"))
+    parser.add_argument("--subrange-incremental", default="false",
+                        help=("Subrange incremental repair (true or false), "
+                              "or use the configured default if not given (false)"))
     parser.add_argument("--datacenters", default=None,
                         help=("a comma separated list of datacenters to repair (do not use spaces after commas). "
                               "Cannot be used in conjunction with --nodes."))
@@ -734,6 +737,7 @@ class ReaperCLI(object):
                                 repairParallelism=args.repair_parallelism,
                                 intensity=args.intensity,
                                 incrementalRepair=args.incremental, 
+                                subrangeIncrementalRepair=args.subrange_incremental,
                                 nodes=args.nodes,
                                 datacenters=args.datacenters,
                                 blacklistedTables=args.blacklisted_tables,
@@ -755,6 +759,7 @@ class ReaperCLI(object):
                                 repairParallelism=args.repair_parallelism,
                                 intensity=args.intensity,
                                 incrementalRepair=args.incremental, 
+                                subrangeIncrementalRepair=args.subrange_incremental,
                                 nodes=args.nodes,
                                 datacenters=args.datacenters,
                                 blacklistedTables=args.blacklisted_tables,
@@ -797,6 +802,7 @@ class ReaperCLI(object):
                                 scheduleDaysBetween=args.schedule_days_between,
                                 scheduleTriggerTime=args.schedule_trigger_time,
                                 incrementalRepair=args.incremental, 
+                                subrangeIncrementalRepair=args.subrange_incremental,
                                 nodes=args.nodes,
                                 datacenters=args.datacenters,
                                 blacklistedTables=args.blacklisted_tables,
@@ -818,6 +824,7 @@ class ReaperCLI(object):
                                 scheduleDaysBetween=args.schedule_days_between,
                                 scheduleTriggerTime=args.schedule_trigger_time,
                                 incrementalRepair=args.incremental, 
+                                subrangeIncrementalRepair=args.subrange_incremental,
                                 nodes=args.nodes,
                                 datacenters=args.datacenters,
                                 blacklistedTables=args.blacklisted_tables,

src/packaging/docker-services/reaper/reaper.env

@@ -35,6 +35,7 @@ REAPER_REPAIR_RUN_THREADS=32
 REAPER_HANGING_REPAIR_TIMEOUT_MINS=90
 REAPER_ENABLE_CROSS_ORIGIN=true
 REAPER_INCREMENTAL_REPAIR=true
+REAPER_SUBRANGE_INCREMENTAL=false
 REAPER_BLACKLIST_TWCS=true
 REAPER_ENABLE_DYNAMIC_SEED_LIST=false
 REAPER_REPAIR_MANAGER_SCHEDULING_INTERVAL_SECONDS=10

src/server/checkstyle.xml

@@ -14,30 +14,11 @@
  limitations under the License.
 -->
 <!DOCTYPE module PUBLIC
-          "-//Puppy Crawl//DTD Check Configuration 1.3//EN"
-          "http://www.puppycrawl.com/dtds/configuration_1_3.dtd">
-
-<!--
-
-   -  Modified from Google Java Style
-      https://google-styleguide.googlecode.com/svn-history/r130/trunk/javaguide.html
-
-   Checkstyle is very configurable. Be sure to read the documentation at
-   http://checkstyle.sf.net (or in your downloaded distribution).
-
-   Most Checks are configurable, be sure to consult the documentation.
-
-   To completely disable a check, just comment it out or delete it from the file.
-
--->
-
-<module name = "Checker">
-    <module name="SuppressWarningsFilter"/>
+        "-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
+        "https://checkstyle.org/dtds/configuration_1_3.dtd">
+<module name = "RootModule">
     <property name="charset" value="UTF-8"/>
-
     <property name="fileExtensions" value="java, properties, xml"/>
-    <!-- Checks for whitespace                               -->
-    <!-- See http://checkstyle.sf.net/config_whitespace.html -->
     <module name="FileTabCharacter">
         <property name="eachLine" value="true"/>
     </module>
@@ -48,9 +29,12 @@
       <property name="multiLines" value="2"/>
     </module>
 
-    <module name="TreeWalker">
-        <module name="SuppressWarningsHolder"/>
+    <module name="LineLength">
+        <property name="max" value="120"/>
+        <property name="ignorePattern" value="^package.*|^import.*|a href|href|http://|https://|ftp://"/>
+    </module>
 
+    <module name="TreeWalker">
         <module name="SuppressWarnings">
             <property name="id" value="checkstyle:suppresswarnings"/>
         </module> 
@@ -70,10 +54,6 @@
             <property name="allowByTailComment" value="true"/>
             <property name="allowNonPrintableEscapes" value="true"/>
         </module>
-        <module name="LineLength">
-            <property name="max" value="120"/>
-            <property name="ignorePattern" value="^package.*|^import.*|a href|href|http://|https://|ftp://"/>
-        </module>
         <module name="Regexp">
           <property name="format" value="[ \t]+$"/>
           <property name="illegalPattern" value="true"/>
@@ -234,6 +214,6 @@
         <module name="FinalClass"/>
     </module>
     <module name="SuppressionFilter">
-      <property name="file" value="${checkstyle.suppressions.file}"/>
+        <property name="file" value="${checkstyle.suppressions.file}"/>
     </module>
 </module>

src/server/src/main/docker/Dockerfile

@@ -27,6 +27,7 @@ ENV REAPER_SEGMENT_COUNT_PER_NODE=64 \
     REAPER_STORAGE_TYPE=memory \
     REAPER_ENABLE_CROSS_ORIGIN=true \
     REAPER_INCREMENTAL_REPAIR=false \
+    REAPER_SUBRANGE_INCREMENTAL=false \
     REAPER_BLACKLIST_TWCS=false \
     REAPER_ENABLE_DYNAMIC_SEED_LIST=true \
     REAPER_REPAIR_MANAGER_SCHEDULING_INTERVAL_SECONDS=30 \

src/server/src/main/docker/cassandra-reaper.yml

@@ -23,6 +23,7 @@ hangingRepairTimeoutMins: ${REAPER_HANGING_REPAIR_TIMEOUT_MINS}
 storageType: ${REAPER_STORAGE_TYPE}
 enableCrossOrigin: ${REAPER_ENABLE_CROSS_ORIGIN}
 incrementalRepair: ${REAPER_INCREMENTAL_REPAIR}
+subrangeIncrementalRepair: ${REAPER_SUBRANGE_INCREMENTAL}
 blacklistTwcsTables: ${REAPER_BLACKLIST_TWCS}
 enableDynamicSeedList: ${REAPER_ENABLE_DYNAMIC_SEED_LIST}
 repairManagerSchedulingIntervalSeconds: ${REAPER_REPAIR_MANAGER_SCHEDULING_INTERVAL_SECONDS}

src/server/src/main/java/io/cassandrareaper/AppContext.java

@@ -56,7 +56,7 @@ public String getLocalNodeAddress() {
     return localNodeAddress;
   }
 
-  private static class Private {
+  private static final class Private {
     private static final Logger LOG = LoggerFactory.getLogger(AppContext.class);
     private static final String DEFAULT_INSTANCE_ADDRESS = "127.0.0.1";
 

src/server/src/main/java/io/cassandrareaper/ReaperApplication.java

@@ -421,6 +421,7 @@ private void schedulePurge(ScheduledExecutorService scheduler) {
   private void checkConfiguration(ReaperApplicationConfiguration config) {
     LOG.debug("repairIntensity: {}", config.getRepairIntensity());
     LOG.debug("incrementalRepair: {}", config.getIncrementalRepair());
+    LOG.debug("subrangeIncrementalRepair: {}", config.getSubrangeIncrementalRepair());
     LOG.debug("repairRunThreadCount: {}", config.getRepairRunThreadCount());
     LOG.debug("segmentCount: {}", config.getSegmentCount());
     LOG.debug("repairParallelism: {}", config.getRepairParallelism());

src/server/src/main/java/io/cassandrareaper/ReaperApplicationConfiguration.java

@@ -74,6 +74,10 @@ public final class ReaperApplicationConfiguration extends Configuration {
   @DefaultValue("false")
   private Boolean incrementalRepair;
 
+  @JsonProperty
+  @DefaultValue("false")
+  private Boolean subrangeIncrementalRepair;
+
   @JsonProperty
   private Boolean blacklistTwcsTables;
 
@@ -237,6 +241,14 @@ public void setIncrementalRepair(boolean incrementalRepair) {
     this.incrementalRepair = incrementalRepair;
   }
 
+  public boolean getSubrangeIncrementalRepair() {
+    return subrangeIncrementalRepair != null ? subrangeIncrementalRepair : false;
+  }
+
+  public void setSubrangeIncrementalRepair(boolean subrangeIncrementalRepair) {
+    this.subrangeIncrementalRepair = subrangeIncrementalRepair;
+  }
+
   public boolean getBlacklistTwcsTables() {
     return blacklistTwcsTables != null ? blacklistTwcsTables : false;
   }
@@ -578,6 +590,9 @@ public static final class AutoSchedulingConfiguration {
     @JsonProperty
     private Boolean incremental;
 
+    @JsonProperty
+    private Boolean subrangeIncrementalRepair;
+
     @JsonProperty
     private Integer percentUnrepairedThreshold;
 
@@ -658,6 +673,14 @@ public void setIncremental(Boolean incremental) {
       this.incremental = incremental;
     }
 
+    public Boolean subrangeIncrementalRepair() {
+      return subrangeIncrementalRepair == null ? false : subrangeIncrementalRepair;
+    }
+
+    public void setSubrangeIncrementalRepair(Boolean subrangeIncrementalRepair) {
+      this.subrangeIncrementalRepair = subrangeIncrementalRepair;
+    }
+
     public Integer getPercentUnrepairedThreshold() {
       return percentUnrepairedThreshold == null ? -1 : percentUnrepairedThreshold;
     }

src/server/src/main/java/io/cassandrareaper/core/RepairType.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2024-2024 DataStax, Inc.
+ *
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.cassandrareaper.core;
+
+public enum RepairType {
+    SUBRANGE_FULL,
+    INCREMENTAL,
+    SUBRANGE_INCREMENTAL;
+
+  public String getRepairType() {
+    return name().toLowerCase();
+  }
+
+  public static RepairType fromName(String name) {
+    return valueOf(name.toUpperCase());
+  }
+
+  public boolean isIncremental() {
+    return this == INCREMENTAL || this == SUBRANGE_INCREMENTAL;
+  }
+
+  public boolean isFull() {
+    return this == SUBRANGE_FULL;
+  }
+
+  public boolean isSubrange() {
+    return this == SUBRANGE_FULL || this == SUBRANGE_INCREMENTAL;
+  }
+}

src/server/src/main/java/io/cassandrareaper/core/RepairUnit.java

@@ -31,6 +31,7 @@ public final class RepairUnit {
   private final String keyspaceName;
   private final Set<String> columnFamilies;
   private final boolean incrementalRepair;
+  private final boolean subrangeIncrementalRepair;
   private final Set<String> nodes;
   private final Set<String> datacenters;
   private final Set<String> blacklistedTables;
@@ -42,7 +43,9 @@ private RepairUnit(Builder builder, UUID id) {
     this.clusterName = builder.clusterName;
     this.keyspaceName = builder.keyspaceName;
     this.columnFamilies = builder.columnFamilies;
-    this.incrementalRepair = builder.incrementalRepair;
+    // If subrange incremental repair is true, we set incremental repair to true as well
+    this.incrementalRepair = builder.incrementalRepair || builder.subrangeIncrementalRepair;
+    this.subrangeIncrementalRepair = builder.subrangeIncrementalRepair;
     this.nodes = builder.nodes;
     this.datacenters = builder.datacenters;
     this.blacklistedTables = builder.blacklistedTables;
@@ -74,6 +77,10 @@ public boolean getIncrementalRepair() {
     return incrementalRepair;
   }
 
+  public boolean getSubrangeIncrementalRepair() {
+    return subrangeIncrementalRepair;
+  }
+
   public Set<String> getNodes() {
     return nodes;
   }
@@ -104,6 +111,7 @@ public static final class Builder {
     public String keyspaceName;
     public Set<String> columnFamilies = Collections.emptySet();
     public Boolean incrementalRepair;
+    public Boolean subrangeIncrementalRepair;
     public Set<String> nodes = Collections.emptySet();
     public Set<String> datacenters = Collections.emptySet();
     public Set<String> blacklistedTables = Collections.emptySet();
@@ -117,6 +125,7 @@ private Builder(RepairUnit original) {
       keyspaceName = original.keyspaceName;
       columnFamilies = original.columnFamilies;
       incrementalRepair = original.incrementalRepair;
+      subrangeIncrementalRepair = original.subrangeIncrementalRepair;
       nodes = original.nodes;
       datacenters = original.datacenters;
       blacklistedTables = original.blacklistedTables;
@@ -144,6 +153,11 @@ public Builder incrementalRepair(boolean incrementalRepair) {
       return this;
     }
 
+    public Builder subrangeIncrementalRepair(boolean subrangeIncrementalRepair) {
+      this.subrangeIncrementalRepair = subrangeIncrementalRepair;
+      return this;
+    }
+
     public Builder nodes(Set<String> nodes) {
       this.nodes = Collections.unmodifiableSet(nodes);
       return this;
@@ -173,6 +187,8 @@ public RepairUnit build(UUID id) {
       Preconditions.checkState(null != clusterName, "clusterName(..) must be called before build(..)");
       Preconditions.checkState(null != keyspaceName, "keyspaceName(..) must be called before build(..)");
       Preconditions.checkState(null != incrementalRepair, "incrementalRepair(..) must be called before build(..)");
+      Preconditions.checkState(null != subrangeIncrementalRepair,
+          "subrangeIncrementalRepair(..) must be called before build(..)");
       Preconditions.checkState(null != repairThreadCount, "repairThreadCount(..) must be called before build(..)");
       Preconditions.checkState(null != timeout, "timeout(..) must be called before build(..)");
       return new RepairUnit(this, id);
@@ -190,6 +206,8 @@ public int hashCode() {
       hash *= 59;
       hash +=  (this.incrementalRepair ? 2 : 1);
       hash *= 59;
+      hash +=  (this.subrangeIncrementalRepair ? 8 : 4);
+      hash *= 59;
       hash +=  Objects.hashCode(this.nodes);
       hash *= 59;
       hash +=  Objects.hashCode(this.datacenters);
@@ -212,6 +230,7 @@ public boolean equals(Object obj) {
       }
 
       return Objects.equals(this.incrementalRepair, ((Builder) obj).incrementalRepair)
+          && Objects.equals(this.subrangeIncrementalRepair, ((Builder) obj).subrangeIncrementalRepair)
           && Objects.equals(this.clusterName, ((Builder) obj).clusterName)
           && Objects.equals(this.keyspaceName, ((Builder) obj).keyspaceName)
           && Objects.equals(this.columnFamilies, ((Builder) obj).columnFamilies)

src/server/src/main/java/io/cassandrareaper/management/ClusterFacade.java

@@ -924,7 +924,7 @@ private Collection<String> enforceLocalNodeForSidecar(Collection<String> endpoin
         : endpoints;
   }
 
-  private static class Async {
+  private static final class Async {
     private static final ExecutorService ASYNC = Executors.newSingleThreadExecutor();
 
     private static boolean markClusterActive(Cluster cluster, AppContext context) {