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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
