You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by st...@apache.org on 2016/07/29 07:34:09 UTC
[01/10] cassandra git commit: Wait for tracing events before
returning response and query at same consistency level client side
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.2 04ef62cba -> 7bd65a129
refs/heads/cassandra-3.0 b7c3dc521 -> 8868b744c
refs/heads/cassandra-3.9 eebadcf1d -> 8e775eaaf
refs/heads/trunk cf68431ad -> bdaa53de4
Wait for tracing events before returning response and query at same consistency level client side
patch by Stefania Alborghetti; reviewed by Paulo Motta for CASSANDRA-11465
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/7bd65a12
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7bd65a12
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7bd65a12
Branch: refs/heads/cassandra-2.2
Commit: 7bd65a129c63091d6885f92afe77a41c4fc46a6f
Parents: 04ef62c
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Jul 14 14:15:39 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:28:03 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ebebbdc..9a16ee3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.2.8
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/bin/cqlsh.py
----------------------------------------------------------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index a3eacdd..b631450 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -1211,7 +1211,7 @@ class Shell(cmd.Cmd):
if self.tracing_enabled:
try:
- for trace in future.get_all_query_traces(self.max_trace_wait):
+ for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level):
print_trace(self, trace)
except TraceUnavailable:
msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index 114795e..343648c 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -112,9 +111,19 @@ public class StageManager
}
}
+ public final static Runnable NO_OP_TASK = new Runnable()
+ {
+ public void run()
+ {
+
+ }
+ };
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
@@ -137,6 +146,11 @@ public class StageManager
@Override
public Future<?> submit(Runnable task)
{
+ if (task.equals(NO_OP_TASK))
+ {
+ assert getMaximumPoolSize() == 1 : "Cannot wait for pending tasks if running more than 1 thread";
+ return super.submit(task);
+ }
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java
index e882e67..03e510f 100644
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Stopwatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.concurrent.Stage;
@@ -36,6 +38,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventNotifier;
@@ -47,6 +50,10 @@ import org.apache.cassandra.utils.progress.ProgressListener;
*/
public class TraceState implements ProgressEventNotifier
{
+ private static final Logger logger = LoggerFactory.getLogger(TraceState.class);
+ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
+ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
+
public final UUID sessionId;
public final InetAddress coordinator;
public final Stopwatch watch;
@@ -119,6 +126,8 @@ public class TraceState implements ProgressEventNotifier
public synchronized void stop()
{
+ waitForPendingEvents();
+
status = Status.STOPPED;
notifyAll();
}
@@ -181,6 +190,8 @@ public class TraceState implements ProgressEventNotifier
final int elapsed = elapsed();
executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
+ if (logger.isTraceEnabled())
+ logger.trace("Adding <{}> to trace events", message);
for (ProgressListener listener : listeners)
{
@@ -194,7 +205,7 @@ public class TraceState implements ProgressEventNotifier
{
protected void runMayThrow() throws Exception
{
- mutateWithCatch(mutation);
+ mutateWithCatch(mutation);
}
});
}
@@ -228,6 +239,33 @@ public class TraceState implements ProgressEventNotifier
}
}
+ /**
+ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
+ * have at least been applied to one replica. This works because the tracking executor only
+ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
+ */
+ protected void waitForPendingEvents()
+ {
+ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
+ return;
+
+ try
+ {
+ if (logger.isTraceEnabled())
+ logger.trace("Waiting for up to {} seconds for trace events to complete",
+ +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
+
+ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
+ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.debug("Failed to wait for tracing events to complete: {}", t);
+ }
+ }
+
+
public boolean acquireReference()
{
while (true)
@@ -242,6 +280,7 @@ public class TraceState implements ProgressEventNotifier
public int releaseReference()
{
+ waitForPendingEvents();
return references.decrementAndGet();
}
}
[04/10] cassandra git commit: Wait for tracing events before
returning response and query at same consistency level client side
Posted by st...@apache.org.
Wait for tracing events before returning response and query at same consistency level client side
patch by Stefania Alborghetti; reviewed by Paulo Motta for CASSANDRA-11465
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/7bd65a12
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7bd65a12
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7bd65a12
Branch: refs/heads/trunk
Commit: 7bd65a129c63091d6885f92afe77a41c4fc46a6f
Parents: 04ef62c
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Jul 14 14:15:39 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:28:03 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ebebbdc..9a16ee3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.2.8
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/bin/cqlsh.py
----------------------------------------------------------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index a3eacdd..b631450 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -1211,7 +1211,7 @@ class Shell(cmd.Cmd):
if self.tracing_enabled:
try:
- for trace in future.get_all_query_traces(self.max_trace_wait):
+ for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level):
print_trace(self, trace)
except TraceUnavailable:
msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index 114795e..343648c 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -112,9 +111,19 @@ public class StageManager
}
}
+ public final static Runnable NO_OP_TASK = new Runnable()
+ {
+ public void run()
+ {
+
+ }
+ };
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
@@ -137,6 +146,11 @@ public class StageManager
@Override
public Future<?> submit(Runnable task)
{
+ if (task.equals(NO_OP_TASK))
+ {
+ assert getMaximumPoolSize() == 1 : "Cannot wait for pending tasks if running more than 1 thread";
+ return super.submit(task);
+ }
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java
index e882e67..03e510f 100644
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Stopwatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.concurrent.Stage;
@@ -36,6 +38,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventNotifier;
@@ -47,6 +50,10 @@ import org.apache.cassandra.utils.progress.ProgressListener;
*/
public class TraceState implements ProgressEventNotifier
{
+ private static final Logger logger = LoggerFactory.getLogger(TraceState.class);
+ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
+ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
+
public final UUID sessionId;
public final InetAddress coordinator;
public final Stopwatch watch;
@@ -119,6 +126,8 @@ public class TraceState implements ProgressEventNotifier
public synchronized void stop()
{
+ waitForPendingEvents();
+
status = Status.STOPPED;
notifyAll();
}
@@ -181,6 +190,8 @@ public class TraceState implements ProgressEventNotifier
final int elapsed = elapsed();
executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
+ if (logger.isTraceEnabled())
+ logger.trace("Adding <{}> to trace events", message);
for (ProgressListener listener : listeners)
{
@@ -194,7 +205,7 @@ public class TraceState implements ProgressEventNotifier
{
protected void runMayThrow() throws Exception
{
- mutateWithCatch(mutation);
+ mutateWithCatch(mutation);
}
});
}
@@ -228,6 +239,33 @@ public class TraceState implements ProgressEventNotifier
}
}
+ /**
+ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
+ * have at least been applied to one replica. This works because the tracking executor only
+ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
+ */
+ protected void waitForPendingEvents()
+ {
+ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
+ return;
+
+ try
+ {
+ if (logger.isTraceEnabled())
+ logger.trace("Waiting for up to {} seconds for trace events to complete",
+ +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
+
+ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
+ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.debug("Failed to wait for tracing events to complete: {}", t);
+ }
+ }
+
+
public boolean acquireReference()
{
while (true)
@@ -242,6 +280,7 @@ public class TraceState implements ProgressEventNotifier
public int releaseReference()
{
+ waitForPendingEvents();
return references.decrementAndGet();
}
}
[07/10] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by st...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8868b744
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8868b744
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8868b744
Branch: refs/heads/trunk
Commit: 8868b744cae5b5b4ef67b7738a117c6e71888235
Parents: b7c3dc5 7bd65a1
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:28:29 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:29:09 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 8f6997b,9a16ee3..bb67fc9
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,29 -1,5 +1,30 @@@
-2.2.8
+3.0.9
+ * Rerun ReplicationAwareTokenAllocatorTest on failure to avoid flakiness (CASSANDRA-12277)
+ * Exception when computing read-repair for range tombstones (CASSANDRA-12263)
+ * Lost counter writes in compact table and static columns (CASSANDRA-12219)
+ * AssertionError with MVs on updating a row that isn't indexed due to a null value (CASSANDRA-12247)
+ * Disable RR and speculative retry with EACH_QUORUM reads (CASSANDRA-11980)
+ * Add option to override compaction space check (CASSANDRA-12180)
+ * Faster startup by only scanning each directory for temporary files once (CASSANDRA-12114)
+ * Respond with v1/v2 protocol header when responding to driver that attempts
+ to connect with too low of a protocol version (CASSANDRA-11464)
+ * NullPointerExpception when reading/compacting table (CASSANDRA-11988)
+ * Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
+ * Fix paging logic for deleted partitions with static columns (CASSANDRA-12107)
+ * Wait until the message is being send to decide which serializer must be used (CASSANDRA-11393)
+ * Fix migration of static thrift column names with non-text comparators (CASSANDRA-12147)
+ * Fix upgrading sparse tables that are incorrectly marked as dense (CASSANDRA-11315)
+ * Fix reverse queries ignoring range tombstones (CASSANDRA-11733)
+ * Avoid potential race when rebuilding CFMetaData (CASSANDRA-12098)
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
+ * Fix column ordering of results with static columns for Thrift requests in
+ a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
+ those static columns in query results (CASSANDRA-12123)
+ * Avoid digest mismatch with empty but static rows (CASSANDRA-12090)
+ * Fix EOF exception when altering column type (CASSANDRA-11820)
+Merged from 2.2:
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
[02/10] cassandra git commit: Wait for tracing events before
returning response and query at same consistency level client side
Posted by st...@apache.org.
Wait for tracing events before returning response and query at same consistency level client side
patch by Stefania Alborghetti; reviewed by Paulo Motta for CASSANDRA-11465
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/7bd65a12
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7bd65a12
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7bd65a12
Branch: refs/heads/cassandra-3.0
Commit: 7bd65a129c63091d6885f92afe77a41c4fc46a6f
Parents: 04ef62c
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Jul 14 14:15:39 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:28:03 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ebebbdc..9a16ee3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.2.8
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/bin/cqlsh.py
----------------------------------------------------------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index a3eacdd..b631450 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -1211,7 +1211,7 @@ class Shell(cmd.Cmd):
if self.tracing_enabled:
try:
- for trace in future.get_all_query_traces(self.max_trace_wait):
+ for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level):
print_trace(self, trace)
except TraceUnavailable:
msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index 114795e..343648c 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -112,9 +111,19 @@ public class StageManager
}
}
+ public final static Runnable NO_OP_TASK = new Runnable()
+ {
+ public void run()
+ {
+
+ }
+ };
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
@@ -137,6 +146,11 @@ public class StageManager
@Override
public Future<?> submit(Runnable task)
{
+ if (task.equals(NO_OP_TASK))
+ {
+ assert getMaximumPoolSize() == 1 : "Cannot wait for pending tasks if running more than 1 thread";
+ return super.submit(task);
+ }
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java
index e882e67..03e510f 100644
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Stopwatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.concurrent.Stage;
@@ -36,6 +38,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventNotifier;
@@ -47,6 +50,10 @@ import org.apache.cassandra.utils.progress.ProgressListener;
*/
public class TraceState implements ProgressEventNotifier
{
+ private static final Logger logger = LoggerFactory.getLogger(TraceState.class);
+ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
+ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
+
public final UUID sessionId;
public final InetAddress coordinator;
public final Stopwatch watch;
@@ -119,6 +126,8 @@ public class TraceState implements ProgressEventNotifier
public synchronized void stop()
{
+ waitForPendingEvents();
+
status = Status.STOPPED;
notifyAll();
}
@@ -181,6 +190,8 @@ public class TraceState implements ProgressEventNotifier
final int elapsed = elapsed();
executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
+ if (logger.isTraceEnabled())
+ logger.trace("Adding <{}> to trace events", message);
for (ProgressListener listener : listeners)
{
@@ -194,7 +205,7 @@ public class TraceState implements ProgressEventNotifier
{
protected void runMayThrow() throws Exception
{
- mutateWithCatch(mutation);
+ mutateWithCatch(mutation);
}
});
}
@@ -228,6 +239,33 @@ public class TraceState implements ProgressEventNotifier
}
}
+ /**
+ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
+ * have at least been applied to one replica. This works because the tracking executor only
+ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
+ */
+ protected void waitForPendingEvents()
+ {
+ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
+ return;
+
+ try
+ {
+ if (logger.isTraceEnabled())
+ logger.trace("Waiting for up to {} seconds for trace events to complete",
+ +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
+
+ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
+ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.debug("Failed to wait for tracing events to complete: {}", t);
+ }
+ }
+
+
public boolean acquireReference()
{
while (true)
@@ -242,6 +280,7 @@ public class TraceState implements ProgressEventNotifier
public int releaseReference()
{
+ waitForPendingEvents();
return references.decrementAndGet();
}
}
[03/10] cassandra git commit: Wait for tracing events before
returning response and query at same consistency level client side
Posted by st...@apache.org.
Wait for tracing events before returning response and query at same consistency level client side
patch by Stefania Alborghetti; reviewed by Paulo Motta for CASSANDRA-11465
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/7bd65a12
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7bd65a12
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7bd65a12
Branch: refs/heads/cassandra-3.9
Commit: 7bd65a129c63091d6885f92afe77a41c4fc46a6f
Parents: 04ef62c
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Jul 14 14:15:39 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:28:03 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ebebbdc..9a16ee3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.2.8
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/bin/cqlsh.py
----------------------------------------------------------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index a3eacdd..b631450 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -1211,7 +1211,7 @@ class Shell(cmd.Cmd):
if self.tracing_enabled:
try:
- for trace in future.get_all_query_traces(self.max_trace_wait):
+ for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level):
print_trace(self, trace)
except TraceUnavailable:
msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index 114795e..343648c 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -112,9 +111,19 @@ public class StageManager
}
}
+ public final static Runnable NO_OP_TASK = new Runnable()
+ {
+ public void run()
+ {
+
+ }
+ };
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
@@ -137,6 +146,11 @@ public class StageManager
@Override
public Future<?> submit(Runnable task)
{
+ if (task.equals(NO_OP_TASK))
+ {
+ assert getMaximumPoolSize() == 1 : "Cannot wait for pending tasks if running more than 1 thread";
+ return super.submit(task);
+ }
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/7bd65a12/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java
index e882e67..03e510f 100644
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Stopwatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.concurrent.Stage;
@@ -36,6 +38,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventNotifier;
@@ -47,6 +50,10 @@ import org.apache.cassandra.utils.progress.ProgressListener;
*/
public class TraceState implements ProgressEventNotifier
{
+ private static final Logger logger = LoggerFactory.getLogger(TraceState.class);
+ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
+ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
+
public final UUID sessionId;
public final InetAddress coordinator;
public final Stopwatch watch;
@@ -119,6 +126,8 @@ public class TraceState implements ProgressEventNotifier
public synchronized void stop()
{
+ waitForPendingEvents();
+
status = Status.STOPPED;
notifyAll();
}
@@ -181,6 +190,8 @@ public class TraceState implements ProgressEventNotifier
final int elapsed = elapsed();
executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
+ if (logger.isTraceEnabled())
+ logger.trace("Adding <{}> to trace events", message);
for (ProgressListener listener : listeners)
{
@@ -194,7 +205,7 @@ public class TraceState implements ProgressEventNotifier
{
protected void runMayThrow() throws Exception
{
- mutateWithCatch(mutation);
+ mutateWithCatch(mutation);
}
});
}
@@ -228,6 +239,33 @@ public class TraceState implements ProgressEventNotifier
}
}
+ /**
+ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
+ * have at least been applied to one replica. This works because the tracking executor only
+ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
+ */
+ protected void waitForPendingEvents()
+ {
+ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
+ return;
+
+ try
+ {
+ if (logger.isTraceEnabled())
+ logger.trace("Waiting for up to {} seconds for trace events to complete",
+ +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
+
+ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
+ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.debug("Failed to wait for tracing events to complete: {}", t);
+ }
+ }
+
+
public boolean acquireReference()
{
while (true)
@@ -242,6 +280,7 @@ public class TraceState implements ProgressEventNotifier
public int releaseReference()
{
+ waitForPendingEvents();
return references.decrementAndGet();
}
}
[06/10] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by st...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8868b744
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8868b744
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8868b744
Branch: refs/heads/cassandra-3.0
Commit: 8868b744cae5b5b4ef67b7738a117c6e71888235
Parents: b7c3dc5 7bd65a1
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:28:29 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:29:09 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 8f6997b,9a16ee3..bb67fc9
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,29 -1,5 +1,30 @@@
-2.2.8
+3.0.9
+ * Rerun ReplicationAwareTokenAllocatorTest on failure to avoid flakiness (CASSANDRA-12277)
+ * Exception when computing read-repair for range tombstones (CASSANDRA-12263)
+ * Lost counter writes in compact table and static columns (CASSANDRA-12219)
+ * AssertionError with MVs on updating a row that isn't indexed due to a null value (CASSANDRA-12247)
+ * Disable RR and speculative retry with EACH_QUORUM reads (CASSANDRA-11980)
+ * Add option to override compaction space check (CASSANDRA-12180)
+ * Faster startup by only scanning each directory for temporary files once (CASSANDRA-12114)
+ * Respond with v1/v2 protocol header when responding to driver that attempts
+ to connect with too low of a protocol version (CASSANDRA-11464)
+ * NullPointerExpception when reading/compacting table (CASSANDRA-11988)
+ * Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
+ * Fix paging logic for deleted partitions with static columns (CASSANDRA-12107)
+ * Wait until the message is being send to decide which serializer must be used (CASSANDRA-11393)
+ * Fix migration of static thrift column names with non-text comparators (CASSANDRA-12147)
+ * Fix upgrading sparse tables that are incorrectly marked as dense (CASSANDRA-11315)
+ * Fix reverse queries ignoring range tombstones (CASSANDRA-11733)
+ * Avoid potential race when rebuilding CFMetaData (CASSANDRA-12098)
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
+ * Fix column ordering of results with static columns for Thrift requests in
+ a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
+ those static columns in query results (CASSANDRA-12123)
+ * Avoid digest mismatch with empty but static rows (CASSANDRA-12090)
+ * Fix EOF exception when altering column type (CASSANDRA-11820)
+Merged from 2.2:
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
[10/10] cassandra git commit: Merge branch 'cassandra-3.9' into trunk
Posted by st...@apache.org.
Merge branch 'cassandra-3.9' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bdaa53de
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bdaa53de
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bdaa53de
Branch: refs/heads/trunk
Commit: bdaa53de4ed938e451cba5cf2df7fa66ef00c153
Parents: cf68431 8e775ea
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:30:57 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:31:21 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 41 ++++----------
.../cassandra/tracing/ExpiredTraceState.java | 10 ++++
.../apache/cassandra/tracing/TraceState.java | 8 +++
.../cassandra/tracing/TraceStateImpl.java | 59 +++++++++++++++++++-
.../apache/cassandra/tracing/TracingImpl.java | 49 ++++++++++++----
.../apache/cassandra/tracing/TracingTest.java | 3 +
8 files changed, 128 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/concurrent/StageManager.java
index badc527,64abf00..84b8da6
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@@ -59,7 -59,7 +59,7 @@@ public class StageManage
stages.put(Stage.TRACING, tracingExecutor());
}
-- private static ExecuteOnlyExecutor tracingExecutor()
++ private static LocalAwareExecutorService tracingExecutor()
{
RejectedExecutionHandler reh = new RejectedExecutionHandler()
{
@@@ -68,13 -68,13 +68,13 @@@
MessagingService.instance().incrementDroppedMessages(MessagingService.Verb._TRACE);
}
};
-- return new ExecuteOnlyExecutor(1,
-- 1,
-- KEEPALIVE,
-- TimeUnit.SECONDS,
-- new ArrayBlockingQueue<Runnable>(1000),
-- new NamedThreadFactory(Stage.TRACING.getJmxName()),
-- reh);
++ return new TracingExecutor(1,
++ 1,
++ KEEPALIVE,
++ TimeUnit.SECONDS,
++ new ArrayBlockingQueue<Runnable>(1000),
++ new NamedThreadFactory(Stage.TRACING.getJmxName()),
++ reh);
}
private static JMXEnabledThreadPoolExecutor multiThreadedStage(Stage stage, int numThreads)
@@@ -112,13 -112,17 +112,12 @@@
}
}
- public final static Runnable NO_OP_TASK = () -> {};
-
/**
-- * A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
- * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
- * a final wait on pending trace events since typically the tracing executor is single-threaded, see
- * CASSANDRA-11465.
++ * The executor used for tracing.
*/
-- private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
++ private static class TracingExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
-- public ExecuteOnlyExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
++ public TracingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
{
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
@@@ -133,23 -137,28 +132,5 @@@
{
execute(command);
}
--
-- @Override
-- public Future<?> submit(Runnable task)
-- {
- if (task.equals(NO_OP_TASK))
- {
- assert getMaximumPoolSize() == 1 : "Cannot wait for pending tasks if running more than 1 thread";
- return super.submit(task);
- }
-- throw new UnsupportedOperationException();
-- }
--
-- @Override
-- public <T> Future<T> submit(Runnable task, T result)
-- {
-- throw new UnsupportedOperationException();
-- }
--
-- @Override
-- public <T> Future<T> submit(Callable<T> task)
-- {
-- throw new UnsupportedOperationException();
-- }
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
index fbe2c33,bc8d5dd..9230d38
--- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
+++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
@@@ -42,4 -42,9 +42,14 @@@ class ExpiredTraceState extends TraceSt
{
delegate.traceImpl(message);
}
+
+ protected void waitForPendingEvents()
+ {
+ delegate.waitForPendingEvents();
+ }
++
++ TraceState getDelegate()
++ {
++ return delegate;
++ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceState.java
index 5365d09,ec2bc9e..b4eff6b
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@@ -179,6 -181,8 +181,11 @@@ public abstract class TraceState implem
protected abstract void traceImpl(String message);
- protected abstract void waitForPendingEvents();
++ protected void waitForPendingEvents()
++ {
++ // if tracing events are asynchronous, then you can use this method to wait for them to complete
++ }
+
public boolean acquireReference()
{
while (true)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceStateImpl.java
index 113ebb7,e2d3a68..55e8389
--- a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
+++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
@@@ -19,7 -19,12 +19,16 @@@ package org.apache.cassandra.tracing
import java.net.InetAddress;
import java.util.Collections;
++import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.ExecutionException;
++import java.util.concurrent.CompletableFuture;
++import java.util.concurrent.ConcurrentHashMap;
++import java.util.concurrent.Future;
+ import java.util.concurrent.TimeUnit;
++import java.util.concurrent.TimeoutException;
+
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
@@@ -35,6 -41,10 +45,12 @@@ import org.apache.cassandra.utils.Wrapp
*/
public class TraceStateImpl extends TraceState
{
+ private static final Logger logger = LoggerFactory.getLogger(TraceStateImpl.class);
+ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
+ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
+
++ private final Set<Future<?>> pendingFutures = ConcurrentHashMap.newKeySet();
++
public TraceStateImpl(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
{
super(coordinator, sessionId, traceType);
@@@ -46,17 -56,45 +62,54 @@@
final int elapsed = elapsed();
executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
+ if (logger.isTraceEnabled())
+ logger.trace("Adding <{}> to trace events", message);
}
- static void executeMutation(final Mutation mutation)
+ /**
- * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
- * have at least been applied to one replica. This works because the tracking executor only
- * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
++ * Wait on submitted futures
+ */
+ protected void waitForPendingEvents()
{
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
+ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
+ return;
+
+ try
+ {
+ if (logger.isTraceEnabled())
- logger.trace("Waiting for up to {} seconds for trace events to complete",
- WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
++ logger.trace("Waiting for up to {} seconds for {} trace events to complete",
++ +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, pendingFutures.size());
+
- StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
- .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
++ CompletableFuture.allOf(pendingFutures.toArray(new CompletableFuture<?>[pendingFutures.size()]))
++ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
++ }
++ catch (TimeoutException ex)
++ {
++ if (logger.isTraceEnabled())
++ logger.trace("Failed to wait for tracing events to complete in {} seconds",
++ WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
- logger.debug("Failed to wait for tracing events to complete: {}", t);
++ logger.error("Got exception whilst waiting for tracing events to complete", t);
+ }
+ }
+
- static void executeMutation(final Mutation mutation)
++
++ void executeMutation(final Mutation mutation)
+ {
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
++ CompletableFuture<Void> fut = CompletableFuture.runAsync(new WrappedRunnable()
{
protected void runMayThrow()
{
mutateWithCatch(mutation);
}
-- });
++ }, StageManager.getStage(Stage.TRACING));
++
++ boolean ret = pendingFutures.add(fut);
++ if (!ret)
++ logger.warn("Failed to insert pending future, tracing synchronization may not work");
}
static void mutateWithCatch(Mutation mutation)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdaa53de/src/java/org/apache/cassandra/tracing/TracingImpl.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TracingImpl.java
index 52ac183,52ac183..4f69584
--- a/src/java/org/apache/cassandra/tracing/TracingImpl.java
+++ b/src/java/org/apache/cassandra/tracing/TracingImpl.java
@@@ -28,10 -28,10 +28,6 @@@ import org.apache.cassandra.concurrent.
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.utils.WrappedRunnable;
--import org.slf4j.Logger;
--import org.slf4j.LoggerFactory;
--
--
/**
* A trace session context. Able to track and store trace sessions. A session is usually a user initiated query, and may
@@@ -39,31 -39,31 +35,62 @@@
*/
class TracingImpl extends Tracing
{
-- private static final Logger logger = LoggerFactory.getLogger(TracingImpl.class);
--
public void stopSessionImpl() {
-- TraceState state = get();
++ final TraceStateImpl state = getStateImpl();
++ if (state == null)
++ return;
++
int elapsed = state.elapsed();
ByteBuffer sessionId = state.sessionIdBytes;
int ttl = state.ttl;
-- TraceStateImpl.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed, ttl));
++
++ state.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed, ttl));
}
public TraceState begin(final String request, final InetAddress client, final Map<String, String> parameters)
{
assert isTracing();
-- final TraceState state = get();
++ final TraceStateImpl state = getStateImpl();
++ assert state != null;
++
final long startedAt = System.currentTimeMillis();
final ByteBuffer sessionId = state.sessionIdBytes;
final String command = state.traceType.toString();
final int ttl = state.ttl;
-- TraceStateImpl.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId, client, parameters, request, startedAt, command, ttl));
--
++ state.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId, client, parameters, request, startedAt, command, ttl));
return state;
}
++ /**
++ * Convert the abstract tracing state to its implementation.
++ *
++ * Expired states are not put in the sessions but the check is for extra safety.
++ *
++ * @return the state converted to its implementation, or null
++ */
++ private TraceStateImpl getStateImpl()
++ {
++ TraceState state = get();
++ if (state == null)
++ return null;
++
++ if (state instanceof ExpiredTraceState)
++ {
++ ExpiredTraceState expiredTraceState = (ExpiredTraceState) state;
++ state = expiredTraceState.getDelegate();
++ }
++
++ if (state instanceof TraceStateImpl)
++ {
++ return (TraceStateImpl)state;
++ }
++
++ assert false : "TracingImpl states should be of type TraceStateImpl";
++ return null;
++ }
++
@Override
protected TraceState newTraceState(InetAddress coordinator, UUID sessionId, TraceType traceType)
{
[08/10] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.9
Posted by st...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.9
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8e775eaa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8e775eaa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8e775eaa
Branch: refs/heads/cassandra-3.9
Commit: 8e775eaaf5ee6b355e050ea15f2e334eff680487
Parents: eebadcf 8868b74
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:29:51 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:30:27 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 12 +++++--
.../cassandra/tracing/ExpiredTraceState.java | 5 +++
.../apache/cassandra/tracing/TraceState.java | 5 +++
.../cassandra/tracing/TraceStateImpl.java | 38 ++++++++++++++++++++
.../apache/cassandra/tracing/TracingTest.java | 3 ++
7 files changed, 63 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index d6a9d16,bb67fc9..ca5b383
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -13,42 -10,6 +13,43 @@@ Merged from 3.0
to connect with too low of a protocol version (CASSANDRA-11464)
* NullPointerExpception when reading/compacting table (CASSANDRA-11988)
* Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
+Merged from 2.2:
++ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
+ * cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
+ * Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
+Merged from 2.1:
+ * cannot use cql since upgrading python to 2.7.11+ (CASSANDRA-11850)
+
+
+3.8
+ * Fix hdr logging for single operation workloads (CASSANDRA-12145)
+ * Fix SASI PREFIX search in CONTAINS mode with partial terms (CASSANDRA-12073)
+ * Increase size of flushExecutor thread pool (CASSANDRA-12071)
+ * Partial revert of CASSANDRA-11971, cannot recycle buffer in SP.sendMessagesToNonlocalDC (CASSANDRA-11950)
+ * Upgrade netty to 4.0.39 (CASSANDRA-12032, CASSANDRA-12034)
+ * Improve details in compaction log message (CASSANDRA-12080)
+ * Allow unset values in CQLSSTableWriter (CASSANDRA-11911)
+ * Chunk cache to request compressor-compatible buffers if pool space is exhausted (CASSANDRA-11993)
+ * Remove DatabaseDescriptor dependencies from SequentialWriter (CASSANDRA-11579)
+ * Move skip_stop_words filter before stemming (CASSANDRA-12078)
+ * Support seek() in EncryptedFileSegmentInputStream (CASSANDRA-11957)
+ * SSTable tools mishandling LocalPartitioner (CASSANDRA-12002)
+ * When SEPWorker assigned work, set thread name to match pool (CASSANDRA-11966)
+ * Add cross-DC latency metrics (CASSANDRA-11596)
+ * Allow terms in selection clause (CASSANDRA-10783)
+ * Add bind variables to trace (CASSANDRA-11719)
+ * Switch counter shards' clock to timestamps (CASSANDRA-9811)
+ * Introduce HdrHistogram and response/service/wait separation to stress tool (CASSANDRA-11853)
+ * entry-weighers in QueryProcessor should respect partitionKeyBindIndexes field (CASSANDRA-11718)
+ * Support older ant versions (CASSANDRA-11807)
+ * Estimate compressed on disk size when deciding if sstable size limit reached (CASSANDRA-11623)
+ * cassandra-stress profiles should support case sensitive schemas (CASSANDRA-11546)
+ * Remove DatabaseDescriptor dependency from FileUtils (CASSANDRA-11578)
+ * Faster streaming (CASSANDRA-9766)
+ * Add prepared query parameter to trace for "Execute CQL3 prepared query" session (CASSANDRA-11425)
+ * Add repaired percentage metric (CASSANDRA-11503)
+ * Add Change-Data-Capture (CASSANDRA-8844)
+Merged from 3.0:
* Fix paging logic for deleted partitions with static columns (CASSANDRA-12107)
* Wait until the message is being send to decide which serializer must be used (CASSANDRA-11393)
* Fix migration of static thrift column names with non-text comparators (CASSANDRA-12147)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/concurrent/StageManager.java
index ebc4f76,a201e78..64abf00
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@@ -113,9 -112,19 +112,13 @@@ public class StageManage
}
}
- public final static Runnable NO_OP_TASK = new Runnable()
- {
- public void run()
- {
-
- }
- };
++ public final static Runnable NO_OP_TASK = () -> {};
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
index fbe2c33,5cc3c21..bc8d5dd
--- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
+++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
@@@ -37,9 -36,4 +37,14 @@@ class ExpiredTraceState extends TraceSt
{
return -1;
}
+
+ protected void traceImpl(String message)
+ {
+ delegate.traceImpl(message);
+ }
++
++ protected void waitForPendingEvents()
++ {
++ delegate.waitForPendingEvents();
++ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceState.java
index 5365d09,03e510f..ec2bc9e
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@@ -177,8 -199,73 +179,10 @@@ public abstract class TraceState implem
}
}
- static void executeMutation(final Mutation mutation)
- {
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
- {
- protected void runMayThrow() throws Exception
- {
- mutateWithCatch(mutation);
- }
- });
- }
-
- /**
- * Called from {@link org.apache.cassandra.net.OutboundTcpConnection} for non-local traces (traces
- * that are not initiated by local node == coordinator).
- */
- public static void mutateWithTracing(final ByteBuffer sessionId, final String message, final int elapsed, final int ttl)
- {
- final String threadName = Thread.currentThread().getName();
-
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
- {
- public void runMayThrow()
- {
- mutateWithCatch(TraceKeyspace.makeEventMutation(sessionId, message, elapsed, threadName, ttl));
- }
- });
- }
-
- static void mutateWithCatch(Mutation mutation)
- {
- try
- {
- StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
- }
- catch (OverloadedException e)
- {
- Tracing.logger.warn("Too many nodes are overloaded to save trace events");
- }
- }
-
- /**
- * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
- * have at least been applied to one replica. This works because the tracking executor only
- * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
- */
- protected void waitForPendingEvents()
- {
- if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
- return;
-
- try
- {
- if (logger.isTraceEnabled())
- logger.trace("Waiting for up to {} seconds for trace events to complete",
- +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
-
- StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
- .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
- }
- catch (Throwable t)
- {
- JVMStabilityInspector.inspectThrowable(t);
- logger.debug("Failed to wait for tracing events to complete: {}", t);
- }
- }
+ protected abstract void traceImpl(String message);
++ protected abstract void waitForPendingEvents();
+
public boolean acquireReference()
{
while (true)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceStateImpl.java
index 113ebb7,0000000..e2d3a68
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
+++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
@@@ -1,74 -1,0 +1,112 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cassandra.tracing;
+
+import java.net.InetAddress;
+import java.util.Collections;
+import java.util.UUID;
++import java.util.concurrent.ExecutionException;
++import java.util.concurrent.TimeUnit;
++
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.service.StorageProxy;
++import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.WrappedRunnable;
+
+/**
+ * ThreadLocal state for a tracing session. The presence of an instance of this class as a ThreadLocal denotes that an
+ * operation is being traced.
+ */
+public class TraceStateImpl extends TraceState
+{
++ private static final Logger logger = LoggerFactory.getLogger(TraceStateImpl.class);
++ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
++ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
++
+ public TraceStateImpl(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
+ {
+ super(coordinator, sessionId, traceType);
+ }
+
+ protected void traceImpl(String message)
+ {
+ final String threadName = Thread.currentThread().getName();
+ final int elapsed = elapsed();
+
+ executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
++ if (logger.isTraceEnabled())
++ logger.trace("Adding <{}> to trace events", message);
++ }
++
++ /**
++ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
++ * have at least been applied to one replica. This works because the tracking executor only
++ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
++ */
++ protected void waitForPendingEvents()
++ {
++ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
++ return;
++
++ try
++ {
++ if (logger.isTraceEnabled())
++ logger.trace("Waiting for up to {} seconds for trace events to complete",
++ WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
++
++ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
++ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
++ }
++ catch (Throwable t)
++ {
++ JVMStabilityInspector.inspectThrowable(t);
++ logger.debug("Failed to wait for tracing events to complete: {}", t);
++ }
+ }
+
+ static void executeMutation(final Mutation mutation)
+ {
+ StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
+ {
+ protected void runMayThrow()
+ {
+ mutateWithCatch(mutation);
+ }
+ });
+ }
+
+ static void mutateWithCatch(Mutation mutation)
+ {
+ try
+ {
+ StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
+ }
+ catch (OverloadedException e)
+ {
+ Tracing.logger.warn("Too many nodes are overloaded to save trace events");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/test/unit/org/apache/cassandra/tracing/TracingTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/tracing/TracingTest.java
index 1b0e507,0000000..30521c0
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/tracing/TracingTest.java
+++ b/test/unit/org/apache/cassandra/tracing/TracingTest.java
@@@ -1,173 -1,0 +1,176 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cassandra.tracing;
+
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.Test;
+
+import org.apache.cassandra.utils.progress.ProgressEvent;
+import org.apache.cassandra.utils.progress.ProgressListener;
+
+public final class TracingTest
+{
+
+ @Test
+ public void test()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.NONE);
+ TraceState state = tracing.begin("test-request", Collections.<String,String>emptyMap());
+ state.trace("test-1");
+ state.trace("test-2");
+ state.trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_get()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.NONE);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().trace("test-1");
+ tracing.get().trace("test-2");
+ tracing.get().trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_get_uuid()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ UUID uuid = tracing.newSession(Tracing.TraceType.NONE);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get(uuid).trace("test-1");
+ tracing.get(uuid).trace("test-2");
+ tracing.get(uuid).trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_states()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.REPAIR);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().enableActivityNotification("test-tag");
+ assert TraceState.Status.IDLE == tracing.get().waitActivity(1);
+ tracing.get().trace("test-1");
+ assert TraceState.Status.ACTIVE == tracing.get().waitActivity(1);
+ tracing.get().stop();
+ assert TraceState.Status.STOPPED == tracing.get().waitActivity(1);
+ tracing.stopSession();
+ assert null == tracing.get();
+ }
+
+ @Test
+ public void test_progress_listener()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.REPAIR);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().enableActivityNotification("test-tag");
+
+ tracing.get().addProgressListener(
+ new ProgressListener()
+ {
+ public void progress(String tag, ProgressEvent pe)
+ {
+ assert "test-tag".equals(tag);
+ assert "test-trace".equals(pe.getMessage());
+ }
+ });
+
+ tracing.get().trace("test-trace");
+ tracing.stopSession();
+ assert null == tracing.get();
+ }
+
+ private class TracingImpl extends Tracing
+ {
+ private final List<String> traces;
+
+ public TracingImpl(List<String> traces)
+ {
+ this.traces = traces;
+ }
+
+ public void stopSessionImpl()
+ {}
+
+ public TraceState begin(String request, InetAddress ia, Map<String, String> map)
+ {
+ traces.add(request);
+ return get();
+ }
+
+ protected TraceState newTraceState(InetAddress ia, UUID uuid, Tracing.TraceType tt)
+ {
+ return new TraceState(ia, uuid, tt)
+ {
+ protected void traceImpl(String string)
+ {
+ traces.add(string);
+ }
+
++ protected void waitForPendingEvents()
++ {
++ }
+ };
+ }
+
+ public void trace(ByteBuffer bb, String string, int i)
+ {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+ }
+}
[05/10] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Posted by st...@apache.org.
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8868b744
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8868b744
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8868b744
Branch: refs/heads/cassandra-3.9
Commit: 8868b744cae5b5b4ef67b7738a117c6e71888235
Parents: b7c3dc5 7bd65a1
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:28:29 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:29:09 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 18 ++++++++-
.../apache/cassandra/tracing/TraceState.java | 41 +++++++++++++++++++-
4 files changed, 58 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 8f6997b,9a16ee3..bb67fc9
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,29 -1,5 +1,30 @@@
-2.2.8
+3.0.9
+ * Rerun ReplicationAwareTokenAllocatorTest on failure to avoid flakiness (CASSANDRA-12277)
+ * Exception when computing read-repair for range tombstones (CASSANDRA-12263)
+ * Lost counter writes in compact table and static columns (CASSANDRA-12219)
+ * AssertionError with MVs on updating a row that isn't indexed due to a null value (CASSANDRA-12247)
+ * Disable RR and speculative retry with EACH_QUORUM reads (CASSANDRA-11980)
+ * Add option to override compaction space check (CASSANDRA-12180)
+ * Faster startup by only scanning each directory for temporary files once (CASSANDRA-12114)
+ * Respond with v1/v2 protocol header when responding to driver that attempts
+ to connect with too low of a protocol version (CASSANDRA-11464)
+ * NullPointerExpception when reading/compacting table (CASSANDRA-11988)
+ * Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
+ * Fix paging logic for deleted partitions with static columns (CASSANDRA-12107)
+ * Wait until the message is being send to decide which serializer must be used (CASSANDRA-11393)
+ * Fix migration of static thrift column names with non-text comparators (CASSANDRA-12147)
+ * Fix upgrading sparse tables that are incorrectly marked as dense (CASSANDRA-11315)
+ * Fix reverse queries ignoring range tombstones (CASSANDRA-11733)
+ * Avoid potential race when rebuilding CFMetaData (CASSANDRA-12098)
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
+ * Fix column ordering of results with static columns for Thrift requests in
+ a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
+ those static columns in query results (CASSANDRA-12123)
+ * Avoid digest mismatch with empty but static rows (CASSANDRA-12090)
+ * Fix EOF exception when altering column type (CASSANDRA-11820)
+Merged from 2.2:
+ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
* cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
* Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
* Synchronize ThriftServer::stop() (CASSANDRA-12105)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8868b744/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
[09/10] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.9
Posted by st...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.9
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8e775eaa
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8e775eaa
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8e775eaa
Branch: refs/heads/trunk
Commit: 8e775eaaf5ee6b355e050ea15f2e334eff680487
Parents: eebadcf 8868b74
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Jul 29 15:29:51 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Jul 29 15:30:27 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
bin/cqlsh.py | 2 +-
.../cassandra/concurrent/StageManager.java | 12 +++++--
.../cassandra/tracing/ExpiredTraceState.java | 5 +++
.../apache/cassandra/tracing/TraceState.java | 5 +++
.../cassandra/tracing/TraceStateImpl.java | 38 ++++++++++++++++++++
.../apache/cassandra/tracing/TracingTest.java | 3 ++
7 files changed, 63 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index d6a9d16,bb67fc9..ca5b383
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -13,42 -10,6 +13,43 @@@ Merged from 3.0
to connect with too low of a protocol version (CASSANDRA-11464)
* NullPointerExpception when reading/compacting table (CASSANDRA-11988)
* Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
+Merged from 2.2:
++ * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465)
+ * cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979)
+ * Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214)
+Merged from 2.1:
+ * cannot use cql since upgrading python to 2.7.11+ (CASSANDRA-11850)
+
+
+3.8
+ * Fix hdr logging for single operation workloads (CASSANDRA-12145)
+ * Fix SASI PREFIX search in CONTAINS mode with partial terms (CASSANDRA-12073)
+ * Increase size of flushExecutor thread pool (CASSANDRA-12071)
+ * Partial revert of CASSANDRA-11971, cannot recycle buffer in SP.sendMessagesToNonlocalDC (CASSANDRA-11950)
+ * Upgrade netty to 4.0.39 (CASSANDRA-12032, CASSANDRA-12034)
+ * Improve details in compaction log message (CASSANDRA-12080)
+ * Allow unset values in CQLSSTableWriter (CASSANDRA-11911)
+ * Chunk cache to request compressor-compatible buffers if pool space is exhausted (CASSANDRA-11993)
+ * Remove DatabaseDescriptor dependencies from SequentialWriter (CASSANDRA-11579)
+ * Move skip_stop_words filter before stemming (CASSANDRA-12078)
+ * Support seek() in EncryptedFileSegmentInputStream (CASSANDRA-11957)
+ * SSTable tools mishandling LocalPartitioner (CASSANDRA-12002)
+ * When SEPWorker assigned work, set thread name to match pool (CASSANDRA-11966)
+ * Add cross-DC latency metrics (CASSANDRA-11596)
+ * Allow terms in selection clause (CASSANDRA-10783)
+ * Add bind variables to trace (CASSANDRA-11719)
+ * Switch counter shards' clock to timestamps (CASSANDRA-9811)
+ * Introduce HdrHistogram and response/service/wait separation to stress tool (CASSANDRA-11853)
+ * entry-weighers in QueryProcessor should respect partitionKeyBindIndexes field (CASSANDRA-11718)
+ * Support older ant versions (CASSANDRA-11807)
+ * Estimate compressed on disk size when deciding if sstable size limit reached (CASSANDRA-11623)
+ * cassandra-stress profiles should support case sensitive schemas (CASSANDRA-11546)
+ * Remove DatabaseDescriptor dependency from FileUtils (CASSANDRA-11578)
+ * Faster streaming (CASSANDRA-9766)
+ * Add prepared query parameter to trace for "Execute CQL3 prepared query" session (CASSANDRA-11425)
+ * Add repaired percentage metric (CASSANDRA-11503)
+ * Add Change-Data-Capture (CASSANDRA-8844)
+Merged from 3.0:
* Fix paging logic for deleted partitions with static columns (CASSANDRA-12107)
* Wait until the message is being send to decide which serializer must be used (CASSANDRA-11393)
* Fix migration of static thrift column names with non-text comparators (CASSANDRA-12147)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/bin/cqlsh.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/concurrent/StageManager.java
index ebc4f76,a201e78..64abf00
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@@ -113,9 -112,19 +112,13 @@@ public class StageManage
}
}
- public final static Runnable NO_OP_TASK = new Runnable()
- {
- public void run()
- {
-
- }
- };
++ public final static Runnable NO_OP_TASK = () -> {};
+
/**
* A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the
- * tracing stage. See CASSANDRA-1123 for background.
+ * tracing stage. See CASSANDRA-1123 for background. We allow submitting NO_OP tasks, to allow
+ * a final wait on pending trace events since typically the tracing executor is single-threaded, see
+ * CASSANDRA-11465.
*/
private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
index fbe2c33,5cc3c21..bc8d5dd
--- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
+++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
@@@ -37,9 -36,4 +37,14 @@@ class ExpiredTraceState extends TraceSt
{
return -1;
}
+
+ protected void traceImpl(String message)
+ {
+ delegate.traceImpl(message);
+ }
++
++ protected void waitForPendingEvents()
++ {
++ delegate.waitForPendingEvents();
++ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/TraceState.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceState.java
index 5365d09,03e510f..ec2bc9e
--- a/src/java/org/apache/cassandra/tracing/TraceState.java
+++ b/src/java/org/apache/cassandra/tracing/TraceState.java
@@@ -177,8 -199,73 +179,10 @@@ public abstract class TraceState implem
}
}
- static void executeMutation(final Mutation mutation)
- {
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
- {
- protected void runMayThrow() throws Exception
- {
- mutateWithCatch(mutation);
- }
- });
- }
-
- /**
- * Called from {@link org.apache.cassandra.net.OutboundTcpConnection} for non-local traces (traces
- * that are not initiated by local node == coordinator).
- */
- public static void mutateWithTracing(final ByteBuffer sessionId, final String message, final int elapsed, final int ttl)
- {
- final String threadName = Thread.currentThread().getName();
-
- StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
- {
- public void runMayThrow()
- {
- mutateWithCatch(TraceKeyspace.makeEventMutation(sessionId, message, elapsed, threadName, ttl));
- }
- });
- }
-
- static void mutateWithCatch(Mutation mutation)
- {
- try
- {
- StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
- }
- catch (OverloadedException e)
- {
- Tracing.logger.warn("Too many nodes are overloaded to save trace events");
- }
- }
-
- /**
- * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
- * have at least been applied to one replica. This works because the tracking executor only
- * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
- */
- protected void waitForPendingEvents()
- {
- if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
- return;
-
- try
- {
- if (logger.isTraceEnabled())
- logger.trace("Waiting for up to {} seconds for trace events to complete",
- +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
-
- StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
- .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
- }
- catch (Throwable t)
- {
- JVMStabilityInspector.inspectThrowable(t);
- logger.debug("Failed to wait for tracing events to complete: {}", t);
- }
- }
+ protected abstract void traceImpl(String message);
++ protected abstract void waitForPendingEvents();
+
public boolean acquireReference()
{
while (true)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tracing/TraceStateImpl.java
index 113ebb7,0000000..e2d3a68
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
+++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java
@@@ -1,74 -1,0 +1,112 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cassandra.tracing;
+
+import java.net.InetAddress;
+import java.util.Collections;
+import java.util.UUID;
++import java.util.concurrent.ExecutionException;
++import java.util.concurrent.TimeUnit;
++
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.service.StorageProxy;
++import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.WrappedRunnable;
+
+/**
+ * ThreadLocal state for a tracing session. The presence of an instance of this class as a ThreadLocal denotes that an
+ * operation is being traced.
+ */
+public class TraceStateImpl extends TraceState
+{
++ private static final Logger logger = LoggerFactory.getLogger(TraceStateImpl.class);
++ private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS =
++ Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1"));
++
+ public TraceStateImpl(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
+ {
+ super(coordinator, sessionId, traceType);
+ }
+
+ protected void traceImpl(String message)
+ {
+ final String threadName = Thread.currentThread().getName();
+ final int elapsed = elapsed();
+
+ executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl));
++ if (logger.isTraceEnabled())
++ logger.trace("Adding <{}> to trace events", message);
++ }
++
++ /**
++ * Post a no-op event to the TRACING stage, so that we can be sure that any previous mutations
++ * have at least been applied to one replica. This works because the tracking executor only
++ * has one thread in its pool, see {@link StageManager#tracingExecutor()}.
++ */
++ protected void waitForPendingEvents()
++ {
++ if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0)
++ return;
++
++ try
++ {
++ if (logger.isTraceEnabled())
++ logger.trace("Waiting for up to {} seconds for trace events to complete",
++ WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS);
++
++ StageManager.getStage(Stage.TRACING).submit(StageManager.NO_OP_TASK)
++ .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS);
++ }
++ catch (Throwable t)
++ {
++ JVMStabilityInspector.inspectThrowable(t);
++ logger.debug("Failed to wait for tracing events to complete: {}", t);
++ }
+ }
+
+ static void executeMutation(final Mutation mutation)
+ {
+ StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable()
+ {
+ protected void runMayThrow()
+ {
+ mutateWithCatch(mutation);
+ }
+ });
+ }
+
+ static void mutateWithCatch(Mutation mutation)
+ {
+ try
+ {
+ StorageProxy.mutate(Collections.singletonList(mutation), ConsistencyLevel.ANY);
+ }
+ catch (OverloadedException e)
+ {
+ Tracing.logger.warn("Too many nodes are overloaded to save trace events");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8e775eaa/test/unit/org/apache/cassandra/tracing/TracingTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/tracing/TracingTest.java
index 1b0e507,0000000..30521c0
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/tracing/TracingTest.java
+++ b/test/unit/org/apache/cassandra/tracing/TracingTest.java
@@@ -1,173 -1,0 +1,176 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cassandra.tracing;
+
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.Test;
+
+import org.apache.cassandra.utils.progress.ProgressEvent;
+import org.apache.cassandra.utils.progress.ProgressListener;
+
+public final class TracingTest
+{
+
+ @Test
+ public void test()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.NONE);
+ TraceState state = tracing.begin("test-request", Collections.<String,String>emptyMap());
+ state.trace("test-1");
+ state.trace("test-2");
+ state.trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_get()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.NONE);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().trace("test-1");
+ tracing.get().trace("test-2");
+ tracing.get().trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_get_uuid()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ UUID uuid = tracing.newSession(Tracing.TraceType.NONE);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get(uuid).trace("test-1");
+ tracing.get(uuid).trace("test-2");
+ tracing.get(uuid).trace("test-3");
+ tracing.stopSession();
+
+ assert null == tracing.get();
+ assert 4 == traces.size();
+ assert "test-request".equals(traces.get(0));
+ assert "test-1".equals(traces.get(1));
+ assert "test-2".equals(traces.get(2));
+ assert "test-3".equals(traces.get(3));
+ }
+
+ @Test
+ public void test_states()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.REPAIR);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().enableActivityNotification("test-tag");
+ assert TraceState.Status.IDLE == tracing.get().waitActivity(1);
+ tracing.get().trace("test-1");
+ assert TraceState.Status.ACTIVE == tracing.get().waitActivity(1);
+ tracing.get().stop();
+ assert TraceState.Status.STOPPED == tracing.get().waitActivity(1);
+ tracing.stopSession();
+ assert null == tracing.get();
+ }
+
+ @Test
+ public void test_progress_listener()
+ {
+ List<String> traces = new ArrayList<>();
+ Tracing tracing = new TracingImpl(traces);
+ tracing.newSession(Tracing.TraceType.REPAIR);
+ tracing.begin("test-request", Collections.<String,String>emptyMap());
+ tracing.get().enableActivityNotification("test-tag");
+
+ tracing.get().addProgressListener(
+ new ProgressListener()
+ {
+ public void progress(String tag, ProgressEvent pe)
+ {
+ assert "test-tag".equals(tag);
+ assert "test-trace".equals(pe.getMessage());
+ }
+ });
+
+ tracing.get().trace("test-trace");
+ tracing.stopSession();
+ assert null == tracing.get();
+ }
+
+ private class TracingImpl extends Tracing
+ {
+ private final List<String> traces;
+
+ public TracingImpl(List<String> traces)
+ {
+ this.traces = traces;
+ }
+
+ public void stopSessionImpl()
+ {}
+
+ public TraceState begin(String request, InetAddress ia, Map<String, String> map)
+ {
+ traces.add(request);
+ return get();
+ }
+
+ protected TraceState newTraceState(InetAddress ia, UUID uuid, Tracing.TraceType tt)
+ {
+ return new TraceState(ia, uuid, tt)
+ {
+ protected void traceImpl(String string)
+ {
+ traces.add(string);
+ }
+
++ protected void waitForPendingEvents()
++ {
++ }
+ };
+ }
+
+ public void trace(ByteBuffer bb, String string, int i)
+ {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+ }
+}