diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index ce6fee3b1813..9d2f51f2ec7b 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.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.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.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.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 @@ private static RowIterator casInternal(String keyspaceName, } // 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 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 @@ private static PartitionIterator readWithConsensusInternal(SinglePartitionReadCo 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 forwarded = CasForwarding.checkAndForwardConsensusReadIfNeeded(group, + consistencyLevel, + alreadyForwarded); + if (forwarded != null) + return forwarded.result; ConsensusAttemptResult lastResult; do @@ -4364,189 +4360,4 @@ public void setClientRequestSizeMetricsEnabled(boolean enabled) 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 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 responseMessage = (Message) 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 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 responseMessage = (Message) 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 e138339b3843..bd99628ecce8 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 @@ */ 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 warnings) { - this(result, null, warnings); + this(materialize(result), null, warnings); } public CasForwardResponse(PartitionIterator result, List 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 warnings) { - this(null, exception, warnings); + this((FilteredPartition) null, exception, warnings); } - private CasForwardResponse(RowIterator result, CassandraException exception, List warnings) + private CasForwardResponse(FilteredPartition result, CassandraException exception, List 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 static class Serializer implements IVersionedSerializer + { + 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 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 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 responseMessage = (Message) 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 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 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 responseMessage = (Message) 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 04df3b4c6436..f8bcaf1fb474 100644 --- a/test/unit/org/apache/cassandra/service/StorageProxyTest.java +++ b/test/unit/org/apache/cassandra/service/StorageProxyTest.java @@ -122,7 +122,6 @@ public void testShouldHintOnExceedingSize() throws Exception }); } - /** * 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 2b819e20f82b..8697e7f0d939 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.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 static void defineSchema() } @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 void testCasForwardRequestWithRemoteClientState() throws Exception } @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 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 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 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 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 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 expectedRows, + List 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 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 twoRowValues() + { + return rowValues(PartitionIterators.singletonIterator(twoRowResult())); + } + + private static List rowValues(PartitionIterator partitions) + { + assertNotNull("Result should not be null", partitions); + + List 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)); } }