This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new a6f62207ba CEP-45: Fix mutation tracking read/write metrics
a6f62207ba is described below
commit a6f62207ba12ea1be40754298849d6fc704dbfc9
Author: Blake Eggleston <[email protected]>
AuthorDate: Tue Oct 20 16:01:38 2026 -0700
CEP-45: Fix mutation tracking read/write metrics
- Fix local tracked read latency reporting
- Fix coordinated tracked write latency reporting
- Refactor StorageProxy#dispatchMutationsWithRetryOnDifferentSystem
Patch by Blake Eggleston; Reviewed by Caleb Rackliffe for CASSANDRA-21793
---
src/java/org/apache/cassandra/db/ReadCommand.java | 2 +-
.../org/apache/cassandra/service/StorageProxy.java | 236 +++++++++++----------
.../service/reads/tracked/PartialTrackedRead.java | 12 +-
.../test/tracking/MutationTrackingMetricsTest.java | 159 ++++++++++++++
4 files changed, 297 insertions(+), 112 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java
b/src/java/org/apache/cassandra/db/ReadCommand.java
index 12917e398a..859a0e452b 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -627,7 +627,7 @@ public abstract class ReadCommand extends AbstractReadQuery
public UnfilteredPartitionIterator
completeTrackedRead(UnfilteredPartitionIterator iterator, PartialTrackedRead
read)
{
- return completeRead(iterator, read.executionController(),
read.searcher(), read.cfs(), read.startTimeNanos());
+ return completeRead(iterator, read.executionController(),
read.searcher(), read.cfs(), nanoTime() - read.beginPhaseNanos());
}
/**
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java
b/src/java/org/apache/cassandra/service/StorageProxy.java
index 70db523926..f9a6841a67 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -1368,36 +1368,13 @@ public class StorageProxy implements StorageProxyMBean
}
else
{
- if (ex instanceof WriteFailureException)
- {
- writeMetrics.failures.mark();
- writeMetricsForLevel(consistencyLevel).failures.mark();
- WriteFailureException fe = (WriteFailureException)ex;
- Tracing.trace("Write failure; received {} of {} required
replies, failed {} requests",
- fe.received, fe.blockFor,
fe.failureReasonByEndpoint.size());
- }
- else
- {
- writeMetrics.timeouts.mark();
- writeMetricsForLevel(consistencyLevel).timeouts.mark();
- WriteTimeoutException te = (WriteTimeoutException)ex;
- Tracing.trace("Write timeout; received {} of {} required
replies", te.received, te.blockFor);
- }
+ markWriteError(ex, consistencyLevel);
throw ex;
}
}
- catch (UnavailableException e)
- {
- writeMetrics.unavailables.mark();
- writeMetricsForLevel(consistencyLevel).unavailables.mark();
- Tracing.trace("Unavailable");
- throw e;
- }
- catch (OverloadedException e)
+ catch (UnavailableException | OverloadedException e)
{
- writeMetrics.unavailables.mark();
- writeMetricsForLevel(consistencyLevel).unavailables.mark();
- Tracing.trace("Overloaded");
+ markWriteError(e, consistencyLevel);
throw e;
}
finally
@@ -1460,36 +1437,13 @@ public class StorageProxy implements StorageProxyMBean
}
else
{
- if (ex instanceof WriteFailureException)
- {
- writeMetrics.failures.mark();
- writeMetricsForLevel(consistencyLevel).failures.mark();
- WriteFailureException fe = (WriteFailureException)ex;
- Tracing.trace("Write failure; received {} of {} required
replies, failed {} requests",
- fe.received, fe.blockFor,
fe.failureReasonByEndpoint.size());
- }
- else
- {
- writeMetrics.timeouts.mark();
- writeMetricsForLevel(consistencyLevel).timeouts.mark();
- WriteTimeoutException te = (WriteTimeoutException)ex;
- Tracing.trace("Write timeout; received {} of {} required
replies", te.received, te.blockFor);
- }
+ markWriteError(ex, consistencyLevel);
throw ex;
}
}
- catch (UnavailableException e)
- {
- writeMetrics.unavailables.mark();
- writeMetricsForLevel(consistencyLevel).unavailables.mark();
- Tracing.trace("Unavailable");
- throw e;
- }
- catch (OverloadedException e)
+ catch (UnavailableException | OverloadedException e)
{
- writeMetrics.unavailables.mark();
- writeMetricsForLevel(consistencyLevel).unavailables.mark();
- Tracing.trace("Overloaded");
+ markWriteError(e, consistencyLevel);
throw e;
}
finally
@@ -1746,15 +1700,7 @@ public class StorageProxy implements StorageProxyMBean
if (accordMutations != null && preserveTimestamps ==
PreserveTimestamp.mixedTimeSource)
checkMixedTimeSourceHandling();
- // Supports batches with multiple tracked mutations
(previously limited to one).
- List<AbstractWriteResponseHandler<?>> trackedHandlers =
trackedMutations != null ? new ArrayList<>(trackedMutations.size()) : null;
- if (trackedMutations != null)
- {
- for (IMutation trackedMutation : trackedMutations)
- {
-
trackedHandlers.add(TrackedWriteRequest.perform(trackedMutation,
consistencyLevel, requestTime));
- }
- }
+ List<AbstractWriteResponseHandler<?>> trackedHandlers =
startTrackedWrites(trackedMutations, consistencyLevel, requestTime);
IAccordResult<TxnResult> accordResult = accordMutations !=
null ? mutateWithAccordAsync(cm, accordMutations, consistencyLevel,
requestTime, preserveTimestamps) : null;
Tracing.trace("Split mutations into tracked {}, Accord {}, and
untracked {}", trackedMutations, accordMutations, untrackedMutations);
@@ -1768,21 +1714,9 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Successfully wrote untracked
mutations");
}
}
- catch (RetryOnDifferentSystemException e)
- {
- writeMetrics.retryDifferentSystem.mark();
-
writeMetricsForLevel(consistencyLevel).retryDifferentSystem.mark();
- logger.debug("Retrying mutations on different system
because some mutations were misrouted according to Cassandra");
- Tracing.trace("Got {} from untracked mutations, will
retry", e);
- continue;
- }
- catch (CoordinatorBehindException e)
+ catch (RetryOnDifferentSystemException |
CoordinatorBehindException e)
{
- writeMetrics.retryCoordinatorBehind.mark();
-
writeMetricsForLevel(consistencyLevel).retryCoordinatorBehind.mark();
-
mutations.forEach(IMutation::clearCachedSerializationsForRetry);
- logger.debug("Retrying mutations now that coordinator has
caught up to cluster metadata");
- Tracing.trace("Got {} from untracked mutations, will
retry", e);
+ markWriteRetry(e, consistencyLevel, mutations, "untracked
mutations");
continue;
}
catch (Exception e)
@@ -1792,14 +1726,7 @@ public class StorageProxy implements StorageProxyMBean
try
{
- if (trackedHandlers != null)
- {
- for (AbstractWriteResponseHandler<?> handler :
trackedHandlers)
- {
- handler.get();
- }
- Tracing.trace("Successfully wrote tracked mutations");
- }
+ awaitTrackedWrites(trackedHandlers, trackedMutations,
untrackedMutations != null, consistencyLevel, requestTime);
}
catch (Exception e)
{
@@ -1842,6 +1769,75 @@ public class StorageProxy implements StorageProxyMBean
}
}
+ /**
+ * Starts the tracked writes without waiting for them. Supports batches
with multiple tracked mutations.
+ *
+ * @return the handlers to pass to {@link #awaitTrackedWrites}, or null if
there are no tracked mutations
+ */
+ @Nullable
+ private static List<AbstractWriteResponseHandler<?>>
startTrackedWrites(@Nullable List<? extends IMutation> trackedMutations,
ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+ {
+ if (trackedMutations == null)
+ return null;
+
+ List<AbstractWriteResponseHandler<?>> handlers = new
ArrayList<>(trackedMutations.size());
+ try
+ {
+ for (IMutation trackedMutation : trackedMutations)
+ handlers.add(TrackedWriteRequest.perform(trackedMutation,
consistencyLevel, requestTime));
+ }
+ catch (RuntimeException e)
+ {
+ // awaitTrackedWrites is not reached if starting the writes fails
+ markWriteError(e, consistencyLevel);
+ recordTrackedWriteLatency(trackedMutations, false,
consistencyLevel, requestTime);
+ throw e;
+ }
+ return handlers;
+ }
+
+ private static void awaitTrackedWrites(@Nullable
List<AbstractWriteResponseHandler<?>> trackedHandlers,
+ @Nullable List<? extends IMutation>
trackedMutations,
+ boolean alsoWritingUntracked,
+ ConsistencyLevel consistencyLevel,
+ Dispatcher.RequestTime requestTime)
+ {
+ if (trackedHandlers == null)
+ return;
+
+ try
+ {
+ for (AbstractWriteResponseHandler<?> handler : trackedHandlers)
+ handler.get();
+ }
+ catch (WriteTimeoutException | WriteFailureException |
UnavailableException | OverloadedException e)
+ {
+ markWriteError(e, consistencyLevel);
+ throw e;
+ }
+ finally
+ {
+ recordTrackedWriteLatency(trackedMutations, alsoWritingUntracked,
consistencyLevel, requestTime);
+ }
+ Tracing.trace("Successfully wrote tracked mutations");
+ }
+
+ /**
+ * Records coordinator write latency for tracked mutations. If untracked
mutations are also written,
+ * mutate() records the request latency in the aggregate write metrics, so
only the per-table
+ * metrics are updated here.
+ */
+ private static void recordTrackedWriteLatency(List<? extends IMutation>
trackedMutations, boolean alsoWritingUntracked, ConsistencyLevel
consistencyLevel, Dispatcher.RequestTime requestTime)
+ {
+ long latency = nanoTime() - requestTime.startedAtNanos();
+ if (!alsoWritingUntracked)
+ {
+ writeMetrics.addNano(latency);
+ writeMetricsForLevel(consistencyLevel).addNano(latency);
+ }
+ updateCoordinatorWriteLatencyTableMetric(trackedMutations, latency);
+ }
+
private static void checkMixedTimeSourceHandling()
{
AccordSpec.MixedTimeSourceHandling handling =
DatabaseDescriptor.getAccord().mixedTimeSourceHandling;
@@ -1885,26 +1881,60 @@ public class StorageProxy implements StorageProxyMBean
{
r.run();
}
- catch (UnavailableException e)
+ catch (UnavailableException | WriteTimeoutException |
WriteFailureException e)
{
- writeMetrics.unavailables.mark();
- writeMetricsForLevel(consistencyLevel).unavailables.mark();
- Tracing.trace("Unavailable");
+ markWriteError(e, consistencyLevel);
throw e;
}
- catch (WriteTimeoutException e)
+ }
+
+ /**
+ * Marks the retry metric for {@code e}, a signal that the write must be
retried, and traces it.
+ * The caller retries. {@code failedPart} names the part of the write that
raised {@code e}.
+ */
+ private static void markWriteRetry(RuntimeException e, ConsistencyLevel
consistencyLevel, List<? extends IMutation> mutations, String failedPart)
+ {
+ if (e instanceof RetryOnDifferentSystemException)
{
- writeMetrics.timeouts.mark();
- writeMetricsForLevel(consistencyLevel).timeouts.mark();
- Tracing.trace("Write timeout; received {} of {} required replies",
e.received, e.blockFor);
- throw e;
+ writeMetrics.retryDifferentSystem.mark();
+ writeMetricsForLevel(consistencyLevel).retryDifferentSystem.mark();
+ logger.debug("Retrying mutations on different system because some
mutations were misrouted");
+ }
+ else
+ {
+ writeMetrics.retryCoordinatorBehind.mark();
+
writeMetricsForLevel(consistencyLevel).retryCoordinatorBehind.mark();
+ mutations.forEach(IMutation::clearCachedSerializationsForRetry);
+ logger.debug("Retrying mutations now that coordinator has caught
up to cluster metadata");
}
- catch (WriteFailureException e)
+ Tracing.trace("Got {} from {}, will retry", e, failedPart);
+ }
+
+ /**
+ * Marks the write error metric that matches {@code e} and traces it.
Callers decide whether to rethrow.
+ */
+ private static void markWriteError(RuntimeException e, ConsistencyLevel
consistencyLevel)
+ {
+ if (e instanceof WriteFailureException)
{
writeMetrics.failures.mark();
writeMetricsForLevel(consistencyLevel).failures.mark();
- Tracing.trace("Write failure; received {} of {} required replies",
e.received, e.blockFor);
- throw e;
+ WriteFailureException fe = (WriteFailureException) e;
+ Tracing.trace("Write failure; received {} of {} required replies,
failed {} requests",
+ fe.received, fe.blockFor,
fe.failureReasonByEndpoint.size());
+ }
+ else if (e instanceof WriteTimeoutException)
+ {
+ writeMetrics.timeouts.mark();
+ writeMetricsForLevel(consistencyLevel).timeouts.mark();
+ WriteTimeoutException te = (WriteTimeoutException) e;
+ Tracing.trace("Write timeout; received {} of {} required replies",
te.received, te.blockFor);
+ }
+ else if (e instanceof UnavailableException || e instanceof
OverloadedException)
+ {
+ writeMetrics.unavailables.mark();
+ writeMetricsForLevel(consistencyLevel).unavailables.mark();
+ Tracing.trace(e instanceof OverloadedException ? "Overloaded" :
"Unavailable");
}
}
@@ -2015,21 +2045,9 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Successfully wrote normal mutations");
}
}
- catch (RetryOnDifferentSystemException e)
- {
- writeMetrics.retryDifferentSystem.mark();
-
writeMetricsForLevel(consistencyLevel).retryDifferentSystem.mark();
- logger.debug("Retrying batch txn on different system
because some mutations were misrouted");
- Tracing.trace("Got {} from normal mutations, will retry",
e);
- continue;
- }
- catch (CoordinatorBehindException e)
+ catch (RetryOnDifferentSystemException |
CoordinatorBehindException e)
{
- writeMetrics.retryCoordinatorBehind.mark();
-
writeMetricsForLevel(consistencyLevel).retryCoordinatorBehind.mark();
-
mutations.forEach(IMutation::clearCachedSerializationsForRetry);
- logger.debug("Retrying batch now that coordinator has
caught up to cluster metadata");
- Tracing.trace("Got {} from normal mutations, will retry",
e);
+ markWriteRetry(e, consistencyLevel, mutations, "normal
mutations");
continue;
}
catch (Exception e)
diff --git
a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
index 58b358bd29..73d53014b0 100644
---
a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
+++
b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
@@ -41,6 +41,8 @@ import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
+import static org.apache.cassandra.utils.Clock.Global.nanoTime;
+
public abstract class PartialTrackedRead
{
private static final Logger logger =
LoggerFactory.getLogger(PartialTrackedRead.class);
@@ -48,6 +50,7 @@ public abstract class PartialTrackedRead
final ReadExecutionController executionController;
final ColumnFamilyStore cfs;
final long startTimeNanos;
+ private volatile long beginPhaseNanos;
public PartialTrackedRead(ReadExecutionController executionController,
ColumnFamilyStore cfs, long startTimeNanos)
{
@@ -66,9 +69,13 @@ public abstract class PartialTrackedRead
return cfs;
}
- public long startTimeNanos()
+ /**
+ * Time spent in the begin phase (sstable selection, index lookups,
initial reads). Local read latency is this
+ * plus the completion phase. The time between the two phases is spent on
reconciliation and is not counted.
+ */
+ public long beginPhaseNanos()
{
- return startTimeNanos;
+ return beginPhaseNanos;
}
abstract ReadCommand command();
@@ -229,6 +236,7 @@ public abstract class PartialTrackedRead
{
logger.trace("Preparing read {}", this);
state = state.asInitialized().prepare(initialData);
+ beginPhaseNanos = nanoTime() - startTimeNanos;
}
void augment(PartitionUpdate update)
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
index 0acdc8f87e..fa37ca0960 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
@@ -19,20 +19,27 @@
package org.apache.cassandra.distributed.test.tracking;
import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.util.concurrent.Uninterruptibles;
import org.awaitility.Awaitility;
import org.junit.Test;
+import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.metrics.ClientRequestsMetricsHolder;
import org.apache.cassandra.metrics.MutationTrackingMetrics;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class MutationTrackingMetricsTest extends TestBaseImpl
{
@@ -354,4 +361,156 @@ public class MutationTrackingMetricsTest extends
TestBaseImpl
.isGreaterThan(initialSpace);
}
}
+
+ /**
+ * The local read latency must not include the time spent waiting for
missing mutations from other replicas.
+ */
+ @Test(timeout = 60000)
+ public void testLocalReadLatencyMetric() throws Throwable
+ {
+ long pullDelayMillis = 2000;
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
+ .set("read_request_timeout", "30000ms"))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ // Node 1 misses the write, so the read has to pull the mutation
from another replica
+ cluster.filters().verbs(Verb.MUTATION_REQ.id).from(2).to(1).drop();
+ cluster.coordinator(2).execute(withKeyspace("INSERT INTO %s.tbl
(pk, val) VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, 1,
"test1");
+ cluster.filters().reset();
+
+ AtomicInteger pulls = new AtomicInteger();
+
cluster.filters().verbs(Verb.MT_PULL_MUTATIONS_REQ.id).from(1).outbound().messagesMatching((from,
to, message) -> {
+ pulls.incrementAndGet();
+ Uninterruptibles.sleepUninterruptibly(pullDelayMillis,
TimeUnit.MILLISECONDS);
+ return false;
+ }).drop();
+
+ long start = System.nanoTime();
+ cluster.coordinator(1).execute(withKeyspace("SELECT * FROM %s.tbl
WHERE pk = ?"),
+ ConsistencyLevel.QUORUM, 1);
+ long elapsedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+ assertThat(pulls.get()).as("The read must wait for a delayed pull
for this test to be meaningful").isGreaterThan(0);
+ assertThat(elapsedMillis).isGreaterThanOrEqualTo(pullDelayMillis);
+
+ long count = cluster.get(1).callOnInstance(() ->
+
Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").metric.readLatency.latency.getCount());
+ long maxMicros = cluster.get(1).callOnInstance(() ->
+
Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").metric.readLatency.latency.getSnapshot().getMax());
+ assertThat(count)
+ .as("Local read latency metric should count tracked reads")
+ .isEqualTo(1);
+ assertThat(maxMicros)
+ .as("Local read latency should not include time spent waiting
for other replicas")
+ .isLessThan(TimeUnit.MILLISECONDS.toMicros(pullDelayMillis) /
2);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testCoordinatorWriteTimeoutMetric() throws Throwable
+ {
+ long timeoutMillis = 1000;
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
+ .set("write_request_timeout",
timeoutMillis + "ms"))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ long initialTimeouts = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.timeouts.getCount());
+ long initialLatencyCount = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.latency.getCount());
+
+ // Only the coordinator's own replica can acknowledge the write
+ cluster.filters().verbs(Verb.MUTATION_REQ.id).from(1).to(2,
3).drop();
+
+ assertThatThrownBy(() ->
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+
ConsistencyLevel.QUORUM, 1, "test1"))
+ .hasMessageContaining("timed out");
+
+ long timeouts = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.timeouts.getCount());
+ long latencyCount = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.latency.getCount());
+ long maxMicros = cluster.get(1).callOnInstance(() ->
+
ClientRequestsMetricsHolder.writeMetrics.latency.getSnapshot().getMax());
+ assertThat(timeouts)
+ .as("Timed out tracked write should be counted")
+ .isEqualTo(initialTimeouts + 1);
+ assertThat(latencyCount)
+ .as("Timed out tracked write should record latency")
+ .isEqualTo(initialLatencyCount + 1);
+ assertThat(maxMicros)
+ .as("Recorded latency should cover the wait for the timeout")
+
.isGreaterThanOrEqualTo(TimeUnit.MILLISECONDS.toMicros(timeoutMillis) / 2);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testCoordinatorWriteLatencyMetric() throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(1)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ long initialCount = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.latency.getCount());
+ long initialTableCount = cluster.get(1).callOnInstance(() ->
+
Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").metric.coordinatorWriteLatency.getCount());
+
+ cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl
(pk, val) VALUES (?, ?)"),
+ ConsistencyLevel.ONE, 1, "test1");
+
+ long afterCount = cluster.get(1).callOnInstance(() ->
+ ClientRequestsMetricsHolder.writeMetrics.latency.getCount());
+ long afterTableCount = cluster.get(1).callOnInstance(() ->
+
Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").metric.coordinatorWriteLatency.getCount());
+ assertThat(afterCount)
+ .as("Coordinator write latency metric should count tracked
writes")
+ .isEqualTo(initialCount + 1);
+ assertThat(afterTableCount)
+ .as("Per-table coordinator write latency metric should count
tracked writes")
+ .isEqualTo(initialTableCount + 1);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testWriteUnavailableMetric() throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(2)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH
replication = " +
+ "{'class': 'SimpleStrategy',
'replication_factor': 2} " +
+ "AND replication_type =
'tracked'"));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+ cluster.get(2).shutdown().get();
+ Awaitility.await().atMost(Duration.ofSeconds(30)).until(() ->
+ cluster.get(1).callOnInstance(() ->
org.apache.cassandra.gms.Gossiper.instance.getLiveMembers().size()) == 1);
+
+ long initial = cluster.get(1).callOnInstance(() ->
+
ClientRequestsMetricsHolder.writeMetrics.unavailables.getCount());
+
+ assertThatThrownBy(() ->
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+
ConsistencyLevel.ALL, 1, "test1"))
+ .hasMessageContaining("Cannot achieve consistency level");
+
+ long after = cluster.get(1).callOnInstance(() ->
+
ClientRequestsMetricsHolder.writeMetrics.unavailables.getCount());
+ assertThat(after)
+ .as("Unavailable write metric should count tracked writes")
+ .isEqualTo(initial + 1);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
