This is an automated email from the ASF dual-hosted git repository.
marcuse pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 6a302f734a1dede4b4bdee5e5f67c5ada6f0115d
Author: Blake Eggleston <bdeggles...@gmail.com>
AuthorDate: Tue Mar 17 16:47:50 2020 +0100
Set now in seconds in the future for validation repairs
Patch by Blake Eggleston; reviewed by Ekaterina Dimitrova for
CASSANDRA-15655
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/config/Config.java | 7 +++++++
.../apache/cassandra/config/DatabaseDescriptor.java | 7 +++++++
src/java/org/apache/cassandra/repair/RepairJob.java | 19 ++++++++++++++++---
4 files changed, 31 insertions(+), 3 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9896272..77d69ca 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0-alpha4
+ * Set now in seconds in the future for validation repairs (CASSANDRA-15655)
* Emit metric on preview repair failure (CASSANDRA-15654)
* Use more appropriate logging levels (CASSANDRA-15661)
* Added production recommendations and improved compaction doc organization
diff --git a/src/java/org/apache/cassandra/config/Config.java
b/src/java/org/apache/cassandra/config/Config.java
index 5a24410..3fc314f 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -471,6 +471,13 @@ public class Config
public volatile boolean report_unconfirmed_repaired_data_mismatches =
false;
/**
+ * number of seconds to set nowInSec into the future when performing
validation previews against repaired data
+ * this (attempts) to prevent a race where validations on different
machines are started on different sides of
+ * a tombstone being compacted away
+ */
+ public volatile int validation_preview_purge_head_start_in_sec = 60 * 60;
+
+ /**
* @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()}
*/
@Deprecated
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 14db023..7af310e 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -3061,4 +3061,11 @@ public class DatabaseDescriptor
throw new ConfigurationException(String.format("%s must be
positive value < %d, but was %d",
name,
unit.overflowThreshold(), val), false);
}
+
+ public static int getValidationPreviewPurgeHeadStartInSec()
+ {
+ int seconds = conf.validation_preview_purge_head_start_in_sec;
+ return Math.max(seconds, 0);
+ }
+
}
diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java
b/src/java/org/apache/cassandra/repair/RepairJob.java
index 3740070..e609f0d 100644
--- a/src/java/org/apache/cassandra/repair/RepairJob.java
+++ b/src/java/org/apache/cassandra/repair/RepairJob.java
@@ -71,6 +71,19 @@ public class RepairJob extends AbstractFuture<RepairResult>
implements Runnable
this.parallelismDegree = session.parallelismDegree;
}
+ public int getNowInSeconds()
+ {
+ int nowInSeconds = FBUtilities.nowInSeconds();
+ if (session.previewKind == PreviewKind.REPAIRED)
+ {
+ return nowInSeconds +
DatabaseDescriptor.getValidationPreviewPurgeHeadStartInSec();
+ }
+ else
+ {
+ return nowInSeconds;
+ }
+ }
+
/**
* Runs repair job.
*
@@ -345,7 +358,7 @@ public class RepairJob extends AbstractFuture<RepairResult>
implements Runnable
String message = String.format("Requesting merkle trees for %s (to
%s)", desc.columnFamily, endpoints);
logger.info("{} {}", session.previewKind.logPrefix(desc.sessionId),
message);
Tracing.traceRepair(message);
- int nowInSec = FBUtilities.nowInSeconds();
+ int nowInSec = getNowInSeconds();
List<ListenableFuture<TreeResponse>> tasks = new
ArrayList<>(endpoints.size());
for (InetAddressAndPort endpoint : endpoints)
{
@@ -365,7 +378,7 @@ public class RepairJob extends AbstractFuture<RepairResult>
implements Runnable
String message = String.format("Requesting merkle trees for %s (to
%s)", desc.columnFamily, endpoints);
logger.info("{} {}", session.previewKind.logPrefix(desc.sessionId),
message);
Tracing.traceRepair(message);
- int nowInSec = FBUtilities.nowInSeconds();
+ int nowInSec = getNowInSeconds();
List<ListenableFuture<TreeResponse>> tasks = new
ArrayList<>(endpoints.size());
Queue<InetAddressAndPort> requests = new LinkedList<>(endpoints);
@@ -407,7 +420,7 @@ public class RepairJob extends AbstractFuture<RepairResult>
implements Runnable
String message = String.format("Requesting merkle trees for %s (to
%s)", desc.columnFamily, endpoints);
logger.info("{} {}", session.previewKind.logPrefix(desc.sessionId),
message);
Tracing.traceRepair(message);
- int nowInSec = FBUtilities.nowInSeconds();
+ int nowInSec = getNowInSeconds();
List<ListenableFuture<TreeResponse>> tasks = new
ArrayList<>(endpoints.size());
Map<String, Queue<InetAddressAndPort>> requestsByDatacenter = new
HashMap<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org
For additional commands, e-mail: commits-h...@cassandra.apache.org
