You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2015/06/04 08:37:15 UTC
cassandra git commit: Make sure repair history actually gets persisted
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.2 49cb12ad1 -> 49c31d683
Make sure repair history actually gets persisted
Patch by marcuse; reviewed by yukim for CASSANDRA-9534
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/49c31d68
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/49c31d68
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/49c31d68
Branch: refs/heads/cassandra-2.2
Commit: 49c31d68376ee9f8c0efaea1579d3d1b6253e16a
Parents: 49cb12a
Author: Marcus Eriksson <ma...@apache.org>
Authored: Wed Jun 3 10:47:42 2015 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Thu Jun 4 08:28:49 2015 +0200
----------------------------------------------------------------------
.../apache/cassandra/cql3/QueryProcessor.java | 7 ++++-
.../repair/SystemDistributedKeyspace.java | 27 +++++++++++++-------
2 files changed, 24 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49c31d68/src/java/org/apache/cassandra/cql3/QueryProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
index 3170932..a071eb9 100644
--- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
@@ -263,7 +263,12 @@ public class QueryProcessor implements QueryHandler
public static UntypedResultSet process(String query, ConsistencyLevel cl) throws RequestExecutionException
{
- ResultMessage result = instance.process(query, QueryState.forInternalCalls(), QueryOptions.forInternalCalls(cl, Collections.<ByteBuffer>emptyList()));
+ return process(query, cl, Collections.<ByteBuffer>emptyList());
+ }
+
+ public static UntypedResultSet process(String query, ConsistencyLevel cl, List<ByteBuffer> values) throws RequestExecutionException
+ {
+ ResultMessage result = instance.process(query, QueryState.forInternalCalls(), QueryOptions.forInternalCalls(cl, values));
if (result instanceof ResultMessage.Rows)
return UntypedResultSet.create(((ResultMessage.Rows)result).result);
else
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49c31d68/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java b/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
index 1969cc5..0f4bde8 100644
--- a/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
+++ b/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
@@ -20,6 +20,8 @@ package org.apache.cassandra.repair;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -36,9 +38,11 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.SimpleStrategy;
+import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
public final class SystemDistributedKeyspace
@@ -102,8 +106,8 @@ public final class SystemDistributedKeyspace
String query = "INSERT INTO %s.%s (parent_id, keyspace_name, columnfamily_names, requested_ranges, started_at)"+
" VALUES (%s, '%s', { '%s' }, { '%s' }, dateOf(now()))";
- String fmtQry = String.format(query, NAME, PARENT_REPAIR_HISTORY, parent_id.toString(), keyspaceName, Joiner.on("','").join(ranges), Joiner.on("','").join(cfnames));
- executeInternalSilent(fmtQry);
+ String fmtQry = String.format(query, NAME, PARENT_REPAIR_HISTORY, parent_id.toString(), keyspaceName, Joiner.on("','").join(cfnames), Joiner.on("','").join(ranges));
+ processSilent(fmtQry);
}
public static void failParentRepair(UUID parent_id, Throwable t)
@@ -114,14 +118,14 @@ public final class SystemDistributedKeyspace
PrintWriter pw = new PrintWriter(sw);
t.printStackTrace(pw);
String fmtQuery = String.format(query, NAME, PARENT_REPAIR_HISTORY, parent_id.toString());
- executeInternalSilent(fmtQuery, t.getMessage(), sw.toString());
+ processSilent(fmtQuery, t.getMessage(), sw.toString());
}
public static void successfulParentRepair(UUID parent_id, Collection<Range<Token>> successfulRanges)
{
String query = "UPDATE %s.%s SET finished_at = dateOf(now()), successful_ranges = {'%s'} WHERE parent_id=%s";
String fmtQuery = String.format(query, NAME, PARENT_REPAIR_HISTORY, Joiner.on("','").join(successfulRanges), parent_id.toString());
- executeInternalSilent(fmtQuery);
+ processSilent(fmtQuery);
}
public static void startRepairs(UUID id, UUID parent_id, String keyspaceName, String[] cfnames, Range<Token> range, Iterable<InetAddress> endpoints)
@@ -148,7 +152,7 @@ public final class SystemDistributedKeyspace
coordinator,
Joiner.on("', '").join(participants),
RepairState.STARTED.toString());
- executeInternalSilent(fmtQry);
+ processSilent(fmtQry);
}
}
@@ -166,7 +170,7 @@ public final class SystemDistributedKeyspace
keyspaceName,
cfname,
id.toString());
- executeInternalSilent(fmtQuery);
+ processSilent(fmtQuery);
}
public static void failedRepairJob(UUID id, String keyspaceName, String cfname, Throwable t)
@@ -180,14 +184,19 @@ public final class SystemDistributedKeyspace
keyspaceName,
cfname,
id.toString());
- executeInternalSilent(fmtQry, t.getMessage(), sw.toString());
+ processSilent(fmtQry, t.getMessage(), sw.toString());
}
- private static void executeInternalSilent(String fmtQry, Object ... values)
+ private static void processSilent(String fmtQry, String... values)
{
try
{
- QueryProcessor.executeInternal(fmtQry, values);
+ List<ByteBuffer> valueList = new ArrayList<>();
+ for (String v : values)
+ {
+ valueList.add(ByteBufferUtil.bytes(v));
+ }
+ QueryProcessor.process(fmtQry, ConsistencyLevel.ONE, valueList);
}
catch (Throwable t)
{