diff --git a/CHANGES.txt b/CHANGES.txt index d6a9d16768..ca5b3831bf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -14,6 +14,7 @@ Merged from 3.0: * 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: diff --git a/bin/cqlsh.py b/bin/cqlsh.py index 2dd561357e..ce85449d50 100644 --- a/bin/cqlsh.py +++ b/bin/cqlsh.py @@ -1218,7 +1218,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,) diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index ebc4f76b5d..64abf00eeb 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.*; @@ -113,9 +112,13 @@ public class StageManager } } + 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 { @@ -138,6 +141,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(); } diff --git a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java index fbe2c331bf..bc8d5dd808 100644 --- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java +++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java @@ -42,4 +42,9 @@ class ExpiredTraceState extends TraceState { delegate.traceImpl(message); } + + protected void waitForPendingEvents() + { + delegate.waitForPendingEvents(); + } } diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java index 5365d0964d..ec2bc9e26b 100644 --- a/src/java/org/apache/cassandra/tracing/TraceState.java +++ b/src/java/org/apache/cassandra/tracing/TraceState.java @@ -111,6 +111,8 @@ public abstract class TraceState implements ProgressEventNotifier public synchronized void stop() { + waitForPendingEvents(); + status = Status.STOPPED; notifyAll(); } @@ -179,6 +181,8 @@ public abstract class TraceState implements ProgressEventNotifier protected abstract void traceImpl(String message); + protected abstract void waitForPendingEvents(); + public boolean acquireReference() { while (true) @@ -193,6 +197,7 @@ public abstract class TraceState implements ProgressEventNotifier public int releaseReference() { + waitForPendingEvents(); return references.decrementAndGet(); } } diff --git a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java index 113ebb7a10..e2d3a68988 100644 --- a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java +++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java @@ -20,6 +20,11 @@ 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; @@ -27,6 +32,7 @@ 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; /** @@ -35,6 +41,10 @@ import org.apache.cassandra.utils.WrappedRunnable; */ 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); @@ -46,6 +56,34 @@ public class TraceStateImpl extends TraceState 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) diff --git a/test/unit/org/apache/cassandra/tracing/TracingTest.java b/test/unit/org/apache/cassandra/tracing/TracingTest.java index 1b0e507270..30521c0bf0 100644 --- a/test/unit/org/apache/cassandra/tracing/TracingTest.java +++ b/test/unit/org/apache/cassandra/tracing/TracingTest.java @@ -162,6 +162,9 @@ public final class TracingTest traces.add(string); } + protected void waitForPendingEvents() + { + } }; }