You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by aw...@apache.org on 2026/08/13 20:16:30 UTC
(cassandra) 01/05: CEP-45 - Fix consensus request forwarding bugs and extract it from StorageProxy
This is an automated email from the ASF dual-hosted git repository.
aweisberg pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 4338fb9d25c98623053b61d125aeafcee4ff2973
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Sun Aug 9 19:49:31 2026 -0700
CEP-45 - Fix consensus request forwarding bugs and extract it from StorageProxy
- CasForwardResponse held a single-use RowIterator that serialize and
serializedSize both drained, so the write pass emitted an empty partition and
tripped messaging's length assertion. deserialize had the mirror defect.
Materialise a FilteredPartition on both sides.
- Forwarded timeouts arrived nested in an ExecutionException and became bare
RuntimeExceptions, erasing the indeterminacy callers judge by exception type.
Examine the cause and translate.
- A CAS that applied reports no result, which read as "no forwarding needed", so
the caller re-ran it locally. Wrap the outcome in Forwarded<T>.
---
.../org/apache/cassandra/service/StorageProxy.java | 211 +----------
.../service/paxos/CasForwardResponse.java | 66 +++-
.../cassandra/service/paxos/CasForwarding.java | 313 ++++++++++++++++
.../apache/cassandra/service/StorageProxyTest.java | 1 -
.../cassandra/service/paxos/CasForwardingTest.java | 395 ++++++++++++++++-----
5 files changed, 672 insertions(+), 314 deletions(-)
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index ce6fee3b18..9d2f51f2ec 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -100,7 +100,6 @@ import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.CasWriteTimeoutException;
import org.apache.cassandra.exceptions.CasWriteUnknownResultException;
-import org.apache.cassandra.exceptions.CassandraException;
import org.apache.cassandra.exceptions.CoordinatorBehindException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.IsBootstrappingException;
@@ -117,7 +116,6 @@ import org.apache.cassandra.exceptions.RetryOnDifferentSystemException;
import org.apache.cassandra.exceptions.UnavailableException;
import org.apache.cassandra.exceptions.WriteFailureException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
-import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.hints.Hint;
import org.apache.cassandra.hints.HintsService;
@@ -175,10 +173,8 @@ import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.C
import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.SplitReads;
import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
import org.apache.cassandra.service.paxos.Ballot;
-import org.apache.cassandra.service.paxos.CasForwardRequest;
-import org.apache.cassandra.service.paxos.CasForwardResponse;
+import org.apache.cassandra.service.paxos.CasForwarding;
import org.apache.cassandra.service.paxos.Commit;
-import org.apache.cassandra.service.paxos.ConsensusReadForwardRequest;
import org.apache.cassandra.service.paxos.ContentionStrategy;
import org.apache.cassandra.service.paxos.Paxos;
import org.apache.cassandra.service.paxos.PaxosCommitForwardRequest;
@@ -230,7 +226,6 @@ import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.writeMetr
import static org.apache.cassandra.net.Message.out;
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.net.Verb.BATCH_STORE_REQ;
-import static org.apache.cassandra.net.Verb.CONSENSUS_READ_FORWARD_REQ;
import static org.apache.cassandra.net.Verb.MUTATION_REQ;
import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_FORWARD_REQ;
import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_REQ;
@@ -272,7 +267,6 @@ public class StorageProxy implements StorageProxyMBean
public static final String UNREACHABLE = "UNREACHABLE";
private static final int FAILURE_LOGGING_INTERVAL_SECONDS = CassandraRelevantProperties.FAILURE_LOGGING_INTERVAL_SECONDS.getInt();
- private static final boolean DISABLE_CONSENSUS_REQUEST_FORWARDING = CassandraRelevantProperties.DISABLE_CONSENSUS_REQUEST_FORWARDING.getBoolean();
private static final String UNSAFE_MIXED_MUTATIONS_MSG = "Mutations look to have different time sources, some are using 'USING TIMESTAMP' and others are using the server timestamp; writes to the Accord table will not be linearizable while using transactions. To allow this behavior set accord.mixed_time_source_handling=log or ignore";
private static final WritePerformer standardWritePerformer;
@@ -441,11 +435,11 @@ public class StorageProxy implements StorageProxyMBean
}
// Check if this CAS operation needs to be forwarded to a replica coordinator for tracked keyspaces
- RowIterator forwardResult = checkAndForwardCasIfNeeded(keyspaceName, cfName, key, request,
- consistencyForPaxos, consistencyForCommit,
- clientState, nowInSeconds, requestTime, alreadyForwarded);
- if (forwardResult != null)
- return forwardResult;
+ CasForwarding.Forwarded<RowIterator> forwarded = CasForwarding.checkAndForwardCasIfNeeded(keyspaceName, cfName, key, request,
+ consistencyForPaxos, consistencyForCommit,
+ clientState, nowInSeconds, alreadyForwarded);
+ if (forwarded != null)
+ return forwarded.result;
ConsensusAttemptResult lastAttemptResult = null;
do
@@ -2732,9 +2726,11 @@ public class StorageProxy implements StorageProxyMBean
throws InvalidRequestException, UnavailableException, ReadFailureException, ReadTimeoutException
{
// Check if this consensus read needs to be forwarded to a replica coordinator for tracked keyspaces
- PartitionIterator forwardResult = checkAndForwardConsensusReadIfNeeded(group, consistencyLevel, requestTime, alreadyForwarded);
- if (forwardResult != null)
- return forwardResult;
+ CasForwarding.Forwarded<PartitionIterator> forwarded = CasForwarding.checkAndForwardConsensusReadIfNeeded(group,
+ consistencyLevel,
+ alreadyForwarded);
+ if (forwarded != null)
+ return forwarded.result;
ConsensusAttemptResult lastResult;
do
@@ -4364,189 +4360,4 @@ public class StorageProxy implements StorageProxyMBean
DatabaseDescriptor.setClientRequestSizeMetricsEnabled(enabled);
}
- /**
- * Check if a CAS operation needs to be forwarded to a replica coordinator for tracked keyspaces.
- * Returns null if no forwarding is needed, or the result of the forwarded operation.
- */
- private static RowIterator checkAndForwardCasIfNeeded(String keyspaceName,
- String cfName,
- DecoratedKey key,
- CQL3CasRequest request,
- ConsistencyLevel consistencyForPaxos,
- ConsistencyLevel consistencyForCommit,
- ClientState clientState,
- long nowInSeconds,
- Dispatcher.RequestTime requestTime,
- boolean alreadyForwarded)
- throws UnavailableException, RequestFailureException, RequestTimeoutException
- {
- Keyspace keyspace = Keyspace.openIfExists(keyspaceName);
- if (keyspace == null)
- throw new KeyspaceNotDefinedException("Keyspace " + keyspaceName + " does not exist");
-
- ClusterMetadata cm = ClusterMetadata.current();
- TableMetadata tableMetadata = cm.schema.getTableMetadata(keyspaceName, cfName);
- if (tableMetadata == null || !MigrationRouter.shouldUseTrackedForWrites(cm, keyspaceName, tableMetadata.id, key.getToken()))
- return null; // Not tracked, no forwarding needed
-
- // Property to disable top-level forwarding for testing
- if (DISABLE_CONSENSUS_REQUEST_FORWARDING)
- return null;
-
- // Check if current coordinator is not a replica
- Token tk = key.getToken();
- EndpointsForToken allReplicas = ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, tk)
- .all();
- EndpointsForToken liveReplicas = allReplicas.filter(FailureDetector.isReplicaAlive);
-
- InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
- boolean isLocalReplica = allReplicas.contains(localEndpoint);
-
- if (isLocalReplica)
- return null; // Local node is a replica, no forwarding needed
-
- // If this request was already forwarded to us and we're not a replica, something is wrong
- if (alreadyForwarded)
- {
- logger.error("Received forwarded CAS for keyspace {} table {} key {} but local node {} is not a replica. Replicas are: {}",
- keyspaceName, cfName, key, localEndpoint, allReplicas);
- Tracing.trace("ERROR: Received forwarded CAS but local node is not a replica");
- throw new InvalidRequestException("Forwarded CAS received by non-replica node " + localEndpoint);
- }
-
- // Find best replica to forward to using proximity-based selection
- if (liveReplicas.isEmpty())
- throw new UnavailableException("No live replicas available for CAS forwarding", consistencyForPaxos, 1, 0);
-
- // Sort by proximity and select the best coordinator
- EndpointsForToken sortedReplicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(localEndpoint, liveReplicas);
- InetAddressAndPort replicaCoordinator = sortedReplicas.get(0).endpoint();
-
- // Create forward request
- CasForwardRequest forwardRequest =
- new CasForwardRequest(keyspaceName, cfName, key, consistencyForPaxos, consistencyForCommit,
- nowInSeconds, clientState, request);
- Message<CasForwardRequest> message = Message.out(Verb.CAS_FORWARD_REQ, forwardRequest);
-
- try
- {
- // Send synchronous request to replica coordinator
- Object responseObj = MessagingService.instance().sendWithResult(message, replicaCoordinator).get();
- @SuppressWarnings("unchecked")
- Message<CasForwardResponse> responseMessage = (Message<CasForwardResponse>) responseObj;
- CasForwardResponse response = responseMessage.payload;
-
- // Add warnings from forwarded operation to local ClientWarn
- for (String warning : response.warnings)
- ClientWarn.instance.warn(warning);
-
- // Check if the forwarded operation had an exception
- if (!response.isSuccess())
- {
- throw response.exception;
- }
-
- return response.result;
- }
- catch (CassandraException ce)
- {
- // Rethrow CassandraExceptions from the replica coordinator
- throw ce;
- }
- catch (Exception e)
- {
- throw new RuntimeException("Failed to forward CAS operation to replica coordinator", e);
- }
- }
-
- /**
- * Check if a consensus read operation needs to be forwarded to a replica coordinator for tracked keyspaces.
- * Returns null if no forwarding is needed, or the result of the forwarded operation.
- */
- private static PartitionIterator checkAndForwardConsensusReadIfNeeded(SinglePartitionReadCommand.Group group,
- ConsistencyLevel consistencyLevel,
- Dispatcher.RequestTime requestTime,
- boolean alreadyForwarded)
- throws UnavailableException, ReadFailureException, ReadTimeoutException
- {
- if (group.queries.isEmpty())
- return null;
-
- // Use the first command to determine keyspace and key for replica planning
- SinglePartitionReadCommand firstCommand = group.queries.get(0);
- String keyspaceName = firstCommand.metadata().keyspace;
-
- Keyspace keyspace = Keyspace.openIfExists(keyspaceName);
- if (keyspace == null)
- throw new KeyspaceNotDefinedException("Keyspace " + keyspaceName + " does not exist");
-
- ClusterMetadata cm = ClusterMetadata.current();
- if (!MigrationRouter.shouldUseTracked(cm, firstCommand))
- return null; // Not tracked, no forwarding needed
-
- // Property to disable top-level forwarding for testing
- if (DISABLE_CONSENSUS_REQUEST_FORWARDING)
- return null;
-
- // Check if current coordinator is not a replica
- Token tk = firstCommand.partitionKey().getToken();
- EndpointsForToken allReplicas = ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, tk)
- .all();
- EndpointsForToken liveReplicas = allReplicas.filter(FailureDetector.isReplicaAlive);
-
- InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
- boolean isLocalReplica = allReplicas.contains(localEndpoint);
-
- if (isLocalReplica)
- return null; // Local node is a replica, no forwarding needed
-
- // If this request was already forwarded to us and we're not a replica, something is wrong
- if (alreadyForwarded)
- {
- logger.error("Received forwarded consensus read for keyspace {} key {} but local node {} is not a replica. Replicas are: {}",
- keyspaceName, firstCommand.partitionKey(), localEndpoint, allReplicas);
- Tracing.trace("ERROR: Received forwarded consensus read but local node is not a replica");
- throw new RuntimeException("Forwarded consensus read received by non-replica node " + localEndpoint);
- }
-
- // Find best replica to forward to using proximity-based selection
- if (liveReplicas.isEmpty())
- throw new UnavailableException("No live replicas available for consensus read forwarding", consistencyLevel, 1, 0);
-
- // Sort by proximity and select the best coordinator
- EndpointsForToken sortedReplicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(localEndpoint, liveReplicas);
- InetAddressAndPort replicaCoordinator = sortedReplicas.get(0).endpoint();
-
- // Create forward request - consensus reads only have a single command
- ConsensusReadForwardRequest forwardRequest = new ConsensusReadForwardRequest(firstCommand, consistencyLevel);
- Message<ConsensusReadForwardRequest> message = Message.out(CONSENSUS_READ_FORWARD_REQ, forwardRequest);
-
- try
- {
- // Send synchronous request to replica coordinator
- Object responseObj = MessagingService.instance().sendWithResult(message, replicaCoordinator).get();
- @SuppressWarnings("unchecked")
- Message<CasForwardResponse> responseMessage = (Message<CasForwardResponse>) responseObj;
- CasForwardResponse response = responseMessage.payload;
-
- // Add warnings from forwarded operation to local ClientWarn
- for (String warning : response.warnings)
- ClientWarn.instance.warn(warning);
-
- // Check if the forwarded operation had an exception
- if (!response.isSuccess())
- throw response.exception;
-
- return response.partitionIterator();
- }
- catch (CassandraException ce)
- {
- // Rethrow CassandraExceptions from the replica coordinator
- throw ce;
- }
- catch (Exception e)
- {
- throw new RuntimeException("Failed to forward consensus read operation to replica coordinator", e);
- }
- }
}
diff --git a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java
index e138339b38..bd99628ecc 100644
--- a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java
+++ b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java
@@ -51,7 +51,7 @@ import static org.apache.cassandra.db.rows.DeserializationHelper.Flag.FROM_REMOT
*/
public class CasForwardResponse
{
- public final RowIterator result;
+ private final FilteredPartition result;
public final CassandraException exception;
@Nonnull
@@ -59,38 +59,67 @@ public class CasForwardResponse
public CasForwardResponse(RowIterator result, List<String> warnings)
{
- this(result, null, warnings);
+ this(materialize(result), null, warnings);
}
public CasForwardResponse(PartitionIterator result, List<String> warnings)
{
- // Extract the single partition from the iterator (consensus reads are single partition)
- this(result != null && result.hasNext() ? result.next() : null, null, warnings);
+ this(materialize(result), null, warnings);
}
public CasForwardResponse(CassandraException exception, List<String> warnings)
{
- this(null, exception, warnings);
+ this((FilteredPartition) null, exception, warnings);
}
- private CasForwardResponse(RowIterator result, CassandraException exception, List<String> warnings)
+ private CasForwardResponse(FilteredPartition result, CassandraException exception, List<String> warnings)
{
this.result = result;
this.exception = exception;
this.warnings = warnings == null ? Collections.emptyList() : warnings;
}
+ private static FilteredPartition materialize(RowIterator rows)
+ {
+ if (rows == null)
+ return null;
+
+ try (RowIterator toClose = rows)
+ {
+ return new FilteredPartition(toClose);
+ }
+ }
+
+ private static FilteredPartition materialize(PartitionIterator partitions)
+ {
+ if (partitions == null)
+ return null;
+
+ try (PartitionIterator toClose = partitions)
+ {
+ return toClose.hasNext() ? materialize(toClose.next()) : null;
+ }
+ }
+
public boolean isSuccess()
{
return exception == null;
}
- /**
- * Get the result as a PartitionIterator.
- */
+ public boolean hasResult()
+ {
+ return result != null;
+ }
+
+ public RowIterator rowIterator()
+ {
+ return result == null ? null : result.rowIterator(false);
+ }
+
public PartitionIterator partitionIterator()
{
- return result == null ? null : PartitionIterators.singletonIterator(result);
+ RowIterator rows = rowIterator();
+ return rows == null ? null : PartitionIterators.singletonIterator(rows);
}
public static final Serializer serializer = new Serializer();
@@ -104,15 +133,15 @@ public class CasForwardResponse
@Override
public void serialize(CasForwardResponse response, DataOutputPlus out, int version) throws IOException
{
- int flags = (response.result != null ? HAS_RESULT : 0)
+ int flags = (response.hasResult() ? HAS_RESULT : 0)
| (response.exception != null ? HAS_EXCEPTION : 0)
| (!response.warnings.isEmpty() ? HAS_WARNINGS : 0)
;
out.write(flags);
- if (response.result != null)
+ if (response.hasResult())
{
- FilteredPartition partition = new FilteredPartition(response.result);
+ FilteredPartition partition = response.result;
partition.metadata().id.serializeCompact(out);
try (UnfilteredRowIterator iterator = partition.unfilteredIterator())
{
@@ -135,14 +164,17 @@ public class CasForwardResponse
boolean hasException = (flags & HAS_EXCEPTION) != 0;
boolean hasWarnings = (flags & HAS_WARNINGS) != 0;
- RowIterator result = null;
+ FilteredPartition result = null;
if (hasResult)
{
TableMetadata metadata = Schema.instance.getExistingTableMetadata(TableId.deserializeCompact(in));
UnfilteredRowIteratorSerializer.Header header = UnfilteredRowIteratorSerializer.serializer.deserializeHeader(metadata, in, version, FROM_REMOTE, STABLE, null);
try (UnfilteredRowIterator partition = UnfilteredRowIteratorSerializer.serializer.deserialize(in, version, metadata, FROM_REMOTE, header))
{
- result = UnfilteredRowIterators.filter(partition, 0);
+ // Materialise inside the block: the deserialized iterator reads lazily from `in`,
+ // so it has to be drained here — both so the result outlives the iterator, and so
+ // the stream is positioned past the partition for the fields that follow.
+ result = new FilteredPartition(UnfilteredRowIterators.filter(partition, 0));
}
}
@@ -162,9 +194,9 @@ public class CasForwardResponse
{
long size = TypeSizes.BYTE_SIZE; // flags byte
- if (response.result != null)
+ if (response.hasResult())
{
- FilteredPartition partition = new FilteredPartition(response.result);
+ FilteredPartition partition = response.result;
size += partition.metadata().id.serializedCompactSize();
try (UnfilteredRowIterator iterator = partition.unfilteredIterator())
{
diff --git a/src/java/org/apache/cassandra/service/paxos/CasForwarding.java b/src/java/org/apache/cassandra/service/paxos/CasForwarding.java
new file mode 100644
index 0000000000..cc388e80f3
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/paxos/CasForwarding.java
@@ -0,0 +1,313 @@
+/*
+ * 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.service.paxos;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.statements.CQL3CasRequest;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.EmptyIterators;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.KeyspaceNotDefinedException;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.db.WriteType;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.CasWriteTimeoutException;
+import org.apache.cassandra.exceptions.CassandraException;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.exceptions.ReadFailureException;
+import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.exceptions.RequestFailureException;
+import org.apache.cassandra.exceptions.RequestFailureReason;
+import org.apache.cassandra.exceptions.RequestTimeoutException;
+import org.apache.cassandra.exceptions.UnavailableException;
+import org.apache.cassandra.exceptions.WriteFailureException;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.locator.EndpointsForToken;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.ReplicaLayout;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.replication.migration.MigrationRouter;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tracing.Tracing;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.net.Verb.CONSENSUS_READ_FORWARD_REQ;
+
+public class CasForwarding
+{
+ private static final Logger logger = LoggerFactory.getLogger(CasForwarding.class);
+ private static final boolean DISABLE_CONSENSUS_REQUEST_FORWARDING = CassandraRelevantProperties.DISABLE_CONSENSUS_REQUEST_FORWARDING.getBoolean();
+
+ /**
+ * Outcome of a forwarding check, distinguishing "no forwarding was needed" from "the request was
+ * forwarded, and this was its result".
+ */
+ public static final class Forwarded<T>
+ {
+ public final T result;
+
+ public Forwarded(T result)
+ {
+ this.result = result;
+ }
+ }
+
+ public static RuntimeException casForwardingFailure(Throwable t, ConsistencyLevel consistencyForPaxos, int blockFor)
+ {
+ MessagingService.FailureResponseException failure = forwardingFailure(t);
+ if (failure == null)
+ return new RuntimeException("Failed to forward CAS operation to replica coordinator", t);
+
+ if (failure.failureReason() == RequestFailureReason.TIMEOUT)
+ return new CasWriteTimeoutException(WriteType.CAS, consistencyForPaxos, 0, blockFor, 0);
+
+ return new WriteFailureException(consistencyForPaxos, 0, blockFor, WriteType.CAS,
+ ImmutableMap.of(failure.from(), failure.failureReason()));
+ }
+
+ public static RuntimeException readForwardingFailure(Throwable t, ConsistencyLevel consistencyLevel, int blockFor)
+ {
+ MessagingService.FailureResponseException failure = forwardingFailure(t);
+ if (failure == null)
+ return new RuntimeException("Failed to forward consensus read operation to replica coordinator", t);
+
+ if (failure.failureReason() == RequestFailureReason.TIMEOUT)
+ return new ReadTimeoutException(consistencyLevel, 0, blockFor, false);
+
+ return new ReadFailureException(consistencyLevel, 0, blockFor, false,
+ ImmutableMap.of(failure.from(), failure.failureReason()));
+ }
+
+ public static MessagingService.FailureResponseException forwardingFailure(Throwable t)
+ {
+ for (Throwable cause = t; cause != null; cause = cause.getCause())
+ {
+ if (cause instanceof MessagingService.FailureResponseException)
+ return (MessagingService.FailureResponseException) cause;
+ }
+ return null;
+ }
+
+ /**
+ * Check if a CAS operation needs to be forwarded to a replica coordinator for tracked keyspaces.
+ * Returns null if no forwarding is needed, otherwise the forwarded operation's result wrapped in a
+ * {@link CasForwarding.Forwarded} — which may hold a null result, since a CAS that applied reports none.
+ */
+ public static CasForwarding.Forwarded<RowIterator> checkAndForwardCasIfNeeded(String keyspaceName,
+ String cfName,
+ DecoratedKey key,
+ CQL3CasRequest request,
+ ConsistencyLevel consistencyForPaxos,
+ ConsistencyLevel consistencyForCommit,
+ ClientState clientState,
+ long nowInSeconds,
+ boolean alreadyForwarded)
+ throws UnavailableException, RequestFailureException, RequestTimeoutException
+ {
+ Keyspace keyspace = Keyspace.openIfExists(keyspaceName);
+ if (keyspace == null)
+ throw new KeyspaceNotDefinedException("Keyspace " + keyspaceName + " does not exist");
+
+ ClusterMetadata cm = ClusterMetadata.current();
+ TableMetadata tableMetadata = cm.schema.getTableMetadata(keyspaceName, cfName);
+ if (tableMetadata == null || !MigrationRouter.shouldUseTrackedForWrites(cm, keyspaceName, tableMetadata.id, key.getToken()))
+ return null; // Not tracked, no forwarding needed
+
+ // Property to disable top-level forwarding for testing
+ if (DISABLE_CONSENSUS_REQUEST_FORWARDING)
+ return null;
+
+ // Check if current coordinator is not a replica
+ Token tk = key.getToken();
+ EndpointsForToken allReplicas = ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, tk)
+ .all();
+ EndpointsForToken liveReplicas = allReplicas.filter(FailureDetector.isReplicaAlive);
+
+ InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+ boolean isLocalReplica = allReplicas.contains(localEndpoint);
+
+ if (isLocalReplica)
+ return null; // Local node is a replica, no forwarding needed
+
+ // If this request was already forwarded to us and we're not a replica, something is wrong
+ if (alreadyForwarded)
+ {
+ logger.error("Received forwarded CAS for keyspace {} table {} key {} but local node {} is not a replica. Replicas are: {}",
+ keyspaceName, cfName, key, localEndpoint, allReplicas);
+ Tracing.trace("ERROR: Received forwarded CAS but local node is not a replica");
+ throw new InvalidRequestException("Forwarded CAS received by non-replica node " + localEndpoint);
+ }
+
+ // Find best replica to forward to using proximity-based selection
+ if (liveReplicas.isEmpty())
+ throw new UnavailableException("No live replicas available for CAS forwarding", consistencyForPaxos, 1, 0);
+
+ // Sort by proximity and select the best coordinator
+ EndpointsForToken sortedReplicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(localEndpoint, liveReplicas);
+ InetAddressAndPort replicaCoordinator = sortedReplicas.get(0).endpoint();
+
+ // Create forward request
+ CasForwardRequest forwardRequest =
+ new CasForwardRequest(keyspaceName, cfName, key, consistencyForPaxos, consistencyForCommit,
+ nowInSeconds, clientState, request);
+ Message<CasForwardRequest> message = Message.out(Verb.CAS_FORWARD_REQ, forwardRequest);
+
+ try
+ {
+ // Send synchronous request to replica coordinator
+ Object responseObj = MessagingService.instance().sendWithResult(message, replicaCoordinator).get();
+ @SuppressWarnings("unchecked")
+ Message<CasForwardResponse> responseMessage = (Message<CasForwardResponse>) responseObj;
+ CasForwardResponse response = responseMessage.payload;
+
+ // Add warnings from forwarded operation to local ClientWarn
+ for (String warning : response.warnings)
+ ClientWarn.instance.warn(warning);
+
+ // Check if the forwarded operation had an exception
+ if (!response.isSuccess())
+ {
+ throw response.exception;
+ }
+
+ // Wrap even when the result is absent: a CAS that applied reports no result, and an
+ // unwrapped null would be indistinguishable from "no forwarding was needed".
+ return new CasForwarding.Forwarded<>(response.rowIterator());
+ }
+ catch (CassandraException ce)
+ {
+ // Rethrow CassandraExceptions from the replica coordinator
+ throw ce;
+ }
+ catch (Exception e)
+ {
+ throw CasForwarding.casForwardingFailure(e, consistencyForPaxos,
+ consistencyForPaxos.blockFor(keyspace.getReplicationStrategy()));
+ }
+ }
+
+ /**
+ * Check if a consensus read operation needs to be forwarded to a replica coordinator for tracked keyspaces.
+ * Returns null if no forwarding is needed, otherwise the forwarded read's result wrapped in a
+ * {@link CasForwarding.Forwarded}.
+ */
+ public static CasForwarding.Forwarded<PartitionIterator> checkAndForwardConsensusReadIfNeeded(SinglePartitionReadCommand.Group group,
+ ConsistencyLevel consistencyLevel,
+ boolean alreadyForwarded)
+ throws UnavailableException, ReadFailureException, ReadTimeoutException
+ {
+ if (group.queries.isEmpty())
+ return null;
+
+ // Use the first command to determine keyspace and key for replica planning
+ SinglePartitionReadCommand firstCommand = group.queries.get(0);
+ String keyspaceName = firstCommand.metadata().keyspace;
+
+ Keyspace keyspace = Keyspace.openIfExists(keyspaceName);
+ if (keyspace == null)
+ throw new KeyspaceNotDefinedException("Keyspace " + keyspaceName + " does not exist");
+
+ ClusterMetadata cm = ClusterMetadata.current();
+ if (!MigrationRouter.shouldUseTracked(cm, firstCommand))
+ return null; // Not tracked, no forwarding needed
+
+ // Property to disable top-level forwarding for testing
+ if (DISABLE_CONSENSUS_REQUEST_FORWARDING)
+ return null;
+
+ // Check if current coordinator is not a replica
+ Token tk = firstCommand.partitionKey().getToken();
+ EndpointsForToken allReplicas = ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, tk)
+ .all();
+ EndpointsForToken liveReplicas = allReplicas.filter(FailureDetector.isReplicaAlive);
+
+ InetAddressAndPort localEndpoint = FBUtilities.getBroadcastAddressAndPort();
+ boolean isLocalReplica = allReplicas.contains(localEndpoint);
+
+ if (isLocalReplica)
+ return null; // Local node is a replica, no forwarding needed
+
+ // If this request was already forwarded to us and we're not a replica, something is wrong
+ if (alreadyForwarded)
+ {
+ logger.error("Received forwarded consensus read for keyspace {} key {} but local node {} is not a replica. Replicas are: {}",
+ keyspaceName, firstCommand.partitionKey(), localEndpoint, allReplicas);
+ Tracing.trace("ERROR: Received forwarded consensus read but local node is not a replica");
+ throw new RuntimeException("Forwarded consensus read received by non-replica node " + localEndpoint);
+ }
+
+ // Find best replica to forward to using proximity-based selection
+ if (liveReplicas.isEmpty())
+ throw new UnavailableException("No live replicas available for consensus read forwarding", consistencyLevel, 1, 0);
+
+ // Sort by proximity and select the best coordinator
+ EndpointsForToken sortedReplicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(localEndpoint, liveReplicas);
+ InetAddressAndPort replicaCoordinator = sortedReplicas.get(0).endpoint();
+
+ // Create forward request - consensus reads only have a single command
+ ConsensusReadForwardRequest forwardRequest = new ConsensusReadForwardRequest(firstCommand, consistencyLevel);
+ Message<ConsensusReadForwardRequest> message = Message.out(CONSENSUS_READ_FORWARD_REQ, forwardRequest);
+
+ try
+ {
+ // Send synchronous request to replica coordinator
+ Object responseObj = MessagingService.instance().sendWithResult(message, replicaCoordinator).get();
+ @SuppressWarnings("unchecked")
+ Message<CasForwardResponse> responseMessage = (Message<CasForwardResponse>) responseObj;
+ CasForwardResponse response = responseMessage.payload;
+
+ // Add warnings from forwarded operation to local ClientWarn
+ for (String warning : response.warnings)
+ ClientWarn.instance.warn(warning);
+
+ // Check if the forwarded operation had an exception
+ if (!response.isSuccess())
+ throw response.exception;
+
+ // An absent result means the forwarded read found no partition. Return an empty iterator
+ // rather than null, which the caller would read as "no forwarding was needed".
+ return new CasForwarding.Forwarded<>(response.hasResult() ? response.partitionIterator()
+ : EmptyIterators.partition());
+ }
+ catch (CassandraException ce)
+ {
+ // Rethrow CassandraExceptions from the replica coordinator
+ throw ce;
+ }
+ catch (Exception e)
+ {
+ throw CasForwarding.readForwardingFailure(e, consistencyLevel,
+ consistencyLevel.blockFor(keyspace.getReplicationStrategy()));
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/service/StorageProxyTest.java b/test/unit/org/apache/cassandra/service/StorageProxyTest.java
index 04df3b4c64..f8bcaf1fb4 100644
--- a/test/unit/org/apache/cassandra/service/StorageProxyTest.java
+++ b/test/unit/org/apache/cassandra/service/StorageProxyTest.java
@@ -122,7 +122,6 @@ public class StorageProxyTest
});
}
-
/**
* Ensure that the timer backing the JMX endpoint to transiently enable blocking read repairs both enables
* and disables the way we'd expect.
diff --git a/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java b/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java
index 2b819e20f8..8697e7f0d9 100644
--- a/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java
+++ b/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java
@@ -17,8 +17,13 @@
*/
package org.apache.cassandra.service.paxos;
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
+import java.util.concurrent.ExecutionException;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -28,17 +33,40 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.CQL3CasRequest;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.RegularAndStaticColumns;
+import org.apache.cassandra.db.RowUpdateBuilder;
+import org.apache.cassandra.db.WriteType;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.PartitionIterators;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.db.rows.UnfilteredRowIterators;
+import org.apache.cassandra.exceptions.CasWriteTimeoutException;
+import org.apache.cassandra.exceptions.ReadFailureException;
+import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.exceptions.RequestFailure;
+import org.apache.cassandra.exceptions.RequestFailureReason;
+import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.exceptions.UnavailableException;
+import org.apache.cassandra.exceptions.WriteFailureException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+import static org.apache.cassandra.service.paxos.CasForwarding.casForwardingFailure;
+import static org.apache.cassandra.service.paxos.CasForwarding.readForwardingFailure;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -50,6 +78,8 @@ public class CasForwardingTest
private static final String KEYSPACE1 = "CasForwardingTest";
private static final String CF_STANDARD1 = "Standard1";
+ private static final int VERSION = MessagingService.current_version;
+
@BeforeClass
public static void defineSchema()
{
@@ -61,46 +91,25 @@ public class CasForwardingTest
}
@Test
- public void testCasForwardResponseExceptionHandling() throws Exception
+ public void testCasForwardRequestWithRemoteClientState()
{
- // Test exception forwarding in CasForwardResponse
- UnavailableException testException = new UnavailableException("Test exception", ConsistencyLevel.QUORUM, 3, 1);
-
- CasForwardResponse response = new CasForwardResponse(testException, null);
-
- assertFalse("Response should not be successful", response.isSuccess());
- assertEquals("Exception should match", testException, response.exception);
- assertNull("Result should be null when exception is present", response.result);
- assertTrue("Warnings should be empty", response.warnings.isEmpty());
- }
-
- @Test
- public void testCasForwardRequestWithRemoteClientState() throws Exception
- {
- // Test CasForwardRequest with RemoteClientState serialization
TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE1, CF_STANDARD1);
DecoratedKey key = DatabaseDescriptor.getPartitioner().decorateKey(ByteBufferUtil.bytes("test"));
ClientState localState = ClientState.forInternalCalls();
localState.setKeyspace(KEYSPACE1);
- // Create a real CQL3CasRequest
CQL3CasRequest casRequest = new CQL3CasRequest(metadata, key, RegularAndStaticColumns.NONE, true, false);
- CasForwardRequest request = new CasForwardRequest(
- KEYSPACE1,
- CF_STANDARD1,
- key,
- ConsistencyLevel.QUORUM,
- ConsistencyLevel.QUORUM,
- System.currentTimeMillis(),
- localState,
- casRequest
- );
-
- // Verify RemoteClientState was created correctly
- assertNotNull("Client state should not be null", request.clientState);
+ CasForwardRequest request = new CasForwardRequest(KEYSPACE1,
+ CF_STANDARD1,
+ key,
+ ConsistencyLevel.QUORUM,
+ ConsistencyLevel.QUORUM,
+ System.currentTimeMillis(),
+ localState,
+ casRequest);
- // Verify other fields
+ assertNotNull("Client state should not be null", request.clientState);
assertEquals("Keyspace name should match", KEYSPACE1, request.keyspaceName);
assertEquals("CF name should match", CF_STANDARD1, request.cfName);
assertEquals("Consistency for paxos should match", ConsistencyLevel.QUORUM, request.consistencyForPaxos);
@@ -109,93 +118,287 @@ public class CasForwardingTest
}
@Test
- public void testCasForwardResponseSerialization() throws Exception
+ public void testExceptionResponse() throws IOException
{
- // Test exception serialization
- UnavailableException testException = new UnavailableException("Test serialization exception", ConsistencyLevel.QUORUM, 3, 1);
- CasForwardResponse originalResponse = new CasForwardResponse(testException, null);
+ UnavailableException exception = new UnavailableException("Test exception", ConsistencyLevel.QUORUM, 3, 1);
+ CasForwardResponse response = new CasForwardResponse(exception, null);
- // Serialize
- DataOutputBuffer out = new DataOutputBuffer();
- CasForwardResponse.serializer.serialize(originalResponse, out, 0);
+ assertFalse("Response should not be successful", response.isSuccess());
+ assertEquals("Exception should match", exception, response.exception);
+ assertFalse("Result should be absent when exception is present", response.hasResult());
+
+ CasForwardResponse deserialized = assertRoundTrips(response, null, Collections.emptyList());
+
+ // UnavailableException rebuilds its message from the consistency level and the required and
+ // alive counts, so compare those rather than the message.
+ UnavailableException deserializedException = (UnavailableException) deserialized.exception;
+ assertEquals("Exception type should match", exception.getClass(), deserializedException.getClass());
+ assertEquals("Consistency level should match", exception.consistency, deserializedException.consistency);
+ assertEquals("Required replicas should match", exception.required, deserializedException.required);
+ assertEquals("Alive replicas should match", exception.alive, deserializedException.alive);
+ }
+
+ @Test
+ public void testNoResultWithWarnings() throws IOException
+ {
+ List<String> warnings = Arrays.asList("Warning 1", "Warning 2", "Test warning message");
+
+ // Both constructors, which is all that distinguishes the CAS verb from the read verb here.
+ assertRoundTrips(new CasForwardResponse((RowIterator) null, warnings), null, warnings);
+ assertRoundTrips(new CasForwardResponse((PartitionIterator) null, warnings), null, warnings);
+ }
- // Deserialize
- DataInputBuffer in = new DataInputBuffer(out.toByteArray());
- CasForwardResponse deserializedResponse = CasForwardResponse.serializer.deserialize(in, 0);
+ @Test
+ public void testResultRoundTrip() throws IOException
+ {
+ assertRoundTrips(new CasForwardResponse(twoRowResult(), null), twoRowValues(), Collections.emptyList());
+ }
- // Verify exception is preserved
- assertFalse("Response should not be successful", deserializedResponse.isSuccess());
- assertNotNull("Exception should not be null", deserializedResponse.exception);
- assertEquals("Exception type should match", testException.getClass(), deserializedResponse.exception.getClass());
+ @Test
+ public void testConsensusReadResultRoundTrip() throws IOException
+ {
+ // The read forwarding verb reaches the same payload through the PartitionIterator constructor.
+ assertRoundTrips(new CasForwardResponse(PartitionIterators.singletonIterator(twoRowResult()), null),
+ twoRowValues(), Collections.emptyList());
+ }
- // Note: UnavailableException reconstructs its message from consistency level, required, and alive values,
- // so we verify the exception type and key properties rather than the exact message
- UnavailableException deserializedException = (UnavailableException) deserializedResponse.exception;
- assertEquals("Consistency level should match", testException.consistency, deserializedException.consistency);
- assertEquals("Required replicas should match", testException.required, deserializedException.required);
- assertEquals("Alive replicas should match", testException.alive, deserializedException.alive);
+ @Test
+ public void testResultAndWarningsRoundTrip() throws IOException
+ {
+ // Warnings follow the result on the wire, so they decode from the wrong offset unless
+ // deserialize reads the result's bytes instead of leaving a lazy view over the stream.
+ List<String> warnings = Arrays.asList("Warning 1", "Warning 2");
+ assertRoundTrips(new CasForwardResponse(twoRowResult(), warnings), twoRowValues(), warnings);
}
@Test
- public void testCasForwardResponseSerializedSizeAccuracy() throws Exception
+ public void testEmptyResultIsDistinctFromNoResult() throws IOException
{
- // Test that serializedSize method returns accurate sizes
- UnavailableException testException = new UnavailableException("Size test exception", ConsistencyLevel.QUORUM, 3, 1);
- CasForwardResponse response = new CasForwardResponse(testException, null);
+ // A successful CAS reports "condition met" as an empty result, which has to stay
+ // distinguishable from carrying no result at all.
+ TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE1, CF_STANDARD1);
+ DecoratedKey key = metadata.partitioner.decorateKey(ByteBufferUtil.bytes("emptyResultKey"));
- // Calculate expected size
- long calculatedSize = CasForwardResponse.serializer.serializedSize(response, 0);
+ assertRoundTrips(new CasForwardResponse(EmptyIterators.row(metadata, key, false), null),
+ Collections.emptyList(), Collections.emptyList());
+ }
- // Serialize and measure actual size
- DataOutputBuffer out = new DataOutputBuffer();
- CasForwardResponse.serializer.serialize(response, out, 0);
+ /**
+ * The result has to be readable an unbounded number of times, through either accessor, and reading
+ * it must not consume it for the reads that follow — including the ones messaging performs.
+ */
+ @Test
+ public void testResultIsReadableRepeatedly() throws IOException
+ {
+ CasForwardResponse response = new CasForwardResponse(twoRowResult(), null);
- long actualSize = out.getLength();
+ List<String> expected = twoRowValues();
+ assertEquals("Fixture should carry two rows", 2, expected.size());
- assertEquals("Calculated size should match actual serialized size", calculatedSize, actualSize);
+ for (int i = 0; i < 3; i++)
+ {
+ assertEquals("partitionIterator() read " + i, expected, rowValues(response.partitionIterator()));
+ assertEquals("rowIterator() read " + i,
+ expected, rowValues(PartitionIterators.singletonIterator(response.rowIterator())));
+ }
+
+ // Reading the result locally, as a handler does, must leave it intact for messaging.
+ assertRoundTrips(response, expected, Collections.emptyList());
}
+ /**
+ * A forwarded operation that timed out is indeterminate — it may have been applied — so it has to
+ * surface as the same timeout type the non-forwarded path raises. Wrapping it in a bare
+ * RuntimeException erases that, and callers (including the simulator's linearizability checker,
+ * which tolerates RequestExecutionException) then treat it as an unexpected internal error.
+ */
@Test
- public void testCasForwardResponseWarningsSerialization() throws Exception
+ public void testCasForwardingTimeoutSurfacesAsCasWriteTimeout() throws UnknownHostException
{
- // Test warnings serialization in CasForwardResponse
- List<String> warnings = Arrays.asList("Warning 1", "Warning 2", "Test warning message");
- CasForwardResponse originalResponse = new CasForwardResponse((RowIterator) null, warnings);
+ RuntimeException translated = casForwardingFailure(forwardingTimeout(),
+ ConsistencyLevel.SERIAL, 2);
+
+ assertThat(translated).isInstanceOf(CasWriteTimeoutException.class)
+ .isInstanceOf(RequestExecutionException.class);
+ CasWriteTimeoutException timeout = (CasWriteTimeoutException) translated;
+ assertThat(timeout.consistency).isEqualTo(ConsistencyLevel.SERIAL);
+ assertThat(timeout.received).isEqualTo(0);
+ assertThat(timeout.blockFor).isEqualTo(2);
+ assertThat(timeout.writeType).isEqualTo(WriteType.CAS);
+ }
- // Serialize
- DataOutputBuffer out = new DataOutputBuffer();
- CasForwardResponse.serializer.serialize(originalResponse, out, 0);
+ @Test
+ public void testConsensusReadForwardingTimeoutSurfacesAsReadTimeout() throws UnknownHostException
+ {
+ RuntimeException translated = readForwardingFailure(forwardingTimeout(), ConsistencyLevel.QUORUM, 2);
+
+ assertThat(translated).isInstanceOf(ReadTimeoutException.class)
+ .isInstanceOf(RequestExecutionException.class);
+ ReadTimeoutException timeout = (ReadTimeoutException) translated;
+ assertThat(timeout.consistency).isEqualTo(ConsistencyLevel.QUORUM);
+ assertThat(timeout.received).isEqualTo(0);
+ assertThat(timeout.blockFor).isEqualTo(2);
+ }
- // Deserialize
- DataInputBuffer in = new DataInputBuffer(out.toByteArray());
- CasForwardResponse deserializedResponse = CasForwardResponse.serializer.deserialize(in, 0);
+ /**
+ * Reasons other than TIMEOUT are definite failures and must not be collapsed into a timeout, which
+ * would tell the caller an operation might have applied when it did not.
+ */
+ @Test
+ public void testCasForwardingNonTimeoutSurfacesAsWriteFailure() throws UnknownHostException
+ {
+ InetAddressAndPort from = InetAddressAndPort.getByName("127.0.0.1:7012");
+ RuntimeException translated = casForwardingFailure(forwardingFailure(from, RequestFailure.INCOMPATIBLE_SCHEMA),
+ ConsistencyLevel.SERIAL, 2);
+
+ assertThat(translated).isInstanceOf(WriteFailureException.class)
+ .isNotInstanceOf(RequestTimeoutException.class);
+ WriteFailureException failure = (WriteFailureException) translated;
+ assertThat(failure.failureReasonByEndpoint)
+ .containsExactly(entry(from, RequestFailureReason.INCOMPATIBLE_SCHEMA));
+ }
- // Verify warnings are preserved
- assertTrue("Response should be successful", deserializedResponse.isSuccess());
- assertFalse("Warnings should not be empty", deserializedResponse.warnings.isEmpty());
- assertEquals("Warnings should match", warnings, deserializedResponse.warnings);
+ @Test
+ public void testConsensusReadForwardingNonTimeoutSurfacesAsReadFailure() throws UnknownHostException
+ {
+ InetAddressAndPort from = InetAddressAndPort.getByName("127.0.0.1:7012");
+ RuntimeException translated = readForwardingFailure(forwardingFailure(from, RequestFailure.UNKNOWN),
+ ConsistencyLevel.QUORUM, 2);
+
+ assertThat(translated).isInstanceOf(ReadFailureException.class)
+ .isNotInstanceOf(RequestTimeoutException.class);
+ ReadFailureException failure = (ReadFailureException) translated;
+ assertThat(failure.failureReasonByEndpoint)
+ .containsExactly(entry(from, RequestFailureReason.UNKNOWN));
}
+ /**
+ * Anything that isn't a failure response has nothing faithful to translate to, so it keeps the
+ * existing wrapper rather than being reported as a timeout.
+ */
@Test
- public void testConsensusReadForwardResponseWarningsSerialization() throws Exception
- {
- // Test warnings serialization in CasForwardResponse (with PartitionIterator constructor)
- List<String> warnings = Arrays.asList("Read warning 1", "Read warning 2");
- CasForwardResponse originalResponse = new CasForwardResponse(
- (org.apache.cassandra.db.partitions.PartitionIterator) null, warnings);
-
- // Serialize
- DataOutputBuffer out = new DataOutputBuffer();
- CasForwardResponse.serializer.serialize(originalResponse, out, 0);
-
- // Deserialize
- DataInputBuffer in = new DataInputBuffer(out.toByteArray());
- CasForwardResponse deserializedResponse = CasForwardResponse.serializer.deserialize(in, 0);
-
- // Verify warnings are preserved
- assertTrue("Response should be successful", deserializedResponse.isSuccess());
- assertNotNull("Warnings should not be null", deserializedResponse.warnings);
- assertEquals("Warning count should match", warnings.size(), deserializedResponse.warnings.size());
- assertEquals("Warnings should match", warnings, deserializedResponse.warnings);
+ public void testForwardingFailureWithoutFailureResponseKeepsGenericWrapper()
+ {
+ IllegalStateException cause = new IllegalStateException("something else went wrong");
+
+ RuntimeException cas = casForwardingFailure(cause, ConsistencyLevel.SERIAL, 2);
+ assertThat(cas).isExactlyInstanceOf(RuntimeException.class)
+ .hasMessage("Failed to forward CAS operation to replica coordinator")
+ .hasCause(cause);
+
+ RuntimeException read = readForwardingFailure(cause, ConsistencyLevel.QUORUM, 2);
+ assertThat(read).isExactlyInstanceOf(RuntimeException.class)
+ .hasMessage("Failed to forward consensus read operation to replica coordinator")
+ .hasCause(cause);
+ }
+
+ private static CasForwardResponse assertRoundTrips(CasForwardResponse response,
+ List<String> expectedRows,
+ List<String> expectedWarnings) throws IOException
+ {
+ assertRows("Result", expectedRows, response);
+
+ byte[] bytes = serializeCheckingSize(response);
+ assertArrayEquals("Repeated serialization should produce identical bytes",
+ bytes, serializeCheckingSize(response));
+
+ CasForwardResponse deserialized;
+ try (DataInputBuffer in = new DataInputBuffer(bytes))
+ {
+ deserialized = CasForwardResponse.serializer.deserialize(in, VERSION);
+ }
+
+ assertEquals("Success should survive the round trip", response.isSuccess(), deserialized.isSuccess());
+ assertEquals("Warnings should survive the round trip", expectedWarnings, deserialized.warnings);
+ assertRows("Deserialized result", expectedRows, deserialized);
+
+ byte[] reserialized = serializeCheckingSize(deserialized);
+ // A deserialized exception picks up the stack frames of the deserialize call, so only the
+ // result path is expected to re-serialize to the same bytes.
+ if (response.exception == null)
+ assertArrayEquals("Re-serializing the deserialized response should produce identical bytes",
+ bytes, reserialized);
+
+ return deserialized;
+ }
+
+ private static byte[] serializeCheckingSize(CasForwardResponse response) throws IOException
+ {
+ long size = CasForwardResponse.serializer.serializedSize(response, VERSION);
+
+ try (DataOutputBuffer out = new DataOutputBuffer())
+ {
+ CasForwardResponse.serializer.serialize(response, out, VERSION);
+ assertEquals("Calculated size should match actual serialized size", size, out.getLength());
+ return out.toByteArray();
+ }
+ }
+
+ private static void assertRows(String what, List<String> expectedRows, CasForwardResponse response)
+ {
+ if (expectedRows == null)
+ {
+ assertFalse(what + " should be absent", response.hasResult());
+ assertNull(what + " should have no partition iterator", response.partitionIterator());
+ assertNull(what + " should have no row iterator", response.rowIterator());
+ }
+ else
+ {
+ assertTrue(what + " should be present", response.hasResult());
+ assertEquals(what + " rows should match", expectedRows, rowValues(response.partitionIterator()));
+ }
+ }
+
+ private static RowIterator twoRowResult()
+ {
+ TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE1, CF_STANDARD1);
+
+ PartitionUpdate update = PartitionUpdate.merge(Arrays.asList(new RowUpdateBuilder(metadata, 1000L, "forwardedKey")
+ .clustering("c1").add("val", "v1")
+ .buildUpdate(),
+ new RowUpdateBuilder(metadata, 1000L, "forwardedKey")
+ .clustering("c2").add("val", "v2")
+ .buildUpdate()));
+
+ return UnfilteredRowIterators.filter(update.unfilteredIterator(), FBUtilities.nowInSeconds());
+ }
+
+ private static List<String> twoRowValues()
+ {
+ return rowValues(PartitionIterators.singletonIterator(twoRowResult()));
+ }
+
+ private static List<String> rowValues(PartitionIterator partitions)
+ {
+ assertNotNull("Result should not be null", partitions);
+
+ List<String> values = new ArrayList<>();
+ try (PartitionIterator iter = partitions)
+ {
+ while (iter.hasNext())
+ {
+ try (RowIterator rows = iter.next())
+ {
+ TableMetadata metadata = rows.metadata();
+ while (rows.hasNext())
+ values.add(rows.next().toString(metadata));
+ }
+ }
+ }
+ return values;
+ }
+
+ /**
+ * The real shape off the wire: the failure response arrives wrapped in an ExecutionException from
+ * the forwarding future, so the translation has to look through the cause chain.
+ */
+ private static Throwable forwardingTimeout() throws UnknownHostException
+ {
+ return forwardingFailure(InetAddressAndPort.getByName("127.0.0.1:7012"), RequestFailure.TIMEOUT);
+ }
+
+ private static Throwable forwardingFailure(InetAddressAndPort from, RequestFailure failure)
+ {
+ return new ExecutionException(new MessagingService.FailureResponseException(from, failure));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org