natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Backport CASSANDRA-17810 fix and improve RTBoundValidator error messages 

patch by Pedro Gordo; reviewed by Josh McKenzie, Stefan Miklosovic for CASSANDRA-18282
This commit is contained in:
Pedro Gordo 2026-04-10 18:49:00 +01:00 committed by Stefan Miklosovic
parent 636409baac
commit 5aec56de3b
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
8 changed files with 227 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.21
* Backport CASSANDRA-17810 fix and improve RTBoundValidator error messages (CASSANDRA-18282)
4.0.20

82
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.db.transform.RTBoundValidator;
import org.apache.cassandra.db.transform.RTBoundValidator.Stage;
import org.apache.cassandra.db.transform.StoppingTransformation;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.QueryCancelledException;
import org.apache.cassandra.exceptions.UnknownIndexException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.IndexNotAvailableException;
@ -391,7 +392,8 @@ public abstract class ReadCommand extends AbstractReadQuery
try
{
iterator = withStateTracking(iterator);
iterator = maybeSlowDownForTesting(iterator);
iterator = withQueryCancellation(iterator);
iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false);
iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos);
@ -557,9 +559,9 @@ public abstract class ReadCommand extends AbstractReadQuery
return Transformation.apply(iter, new MetricRecording());
}
protected class CheckForAbort extends StoppingTransformation<UnfilteredRowIterator>
private class QueryCancellationChecker extends StoppingTransformation<UnfilteredRowIterator>
{
long lastChecked = 0;
long lastCheckedAt = 0;
@Override
protected void attachTo(BasePartitions partitions)
@ -577,57 +579,71 @@ public abstract class ReadCommand extends AbstractReadQuery
this.rows = rows;
}
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
if (maybeAbort())
{
partition.close();
return null;
}
maybeCancel();
return Transformation.apply(partition, this);
}
@Override
protected Row applyToRow(Row row)
{
if (TEST_ITERATION_DELAY_MILLIS > 0)
maybeDelayForTesting();
return maybeAbort() ? null : row;
maybeCancel();
return row;
}
private boolean maybeAbort()
private void maybeCancel()
{
/**
* TODO: this is not a great way to abort early; why not expressly limit checks to 10ms intervals?
/*
* The value returned by approxTime.now() is updated only every
* {@link org.apache.cassandra.utils.MonotonicClock.SampledClock.CHECK_INTERVAL_MS}, by default 2 millis. Since MonitorableImpl
* relies on approxTime, we don't need to check unless the approximate time has elapsed.
* {@link org.apache.cassandra.utils.MonotonicClock.SampledClock.CHECK_INTERVAL_MS}, by default 2 millis.
* Since MonitorableImpl relies on approxTime, we don't need to check unless the approximate time has elapsed.
*/
if (lastChecked == approxTime.now())
return false;
lastChecked = approxTime.now();
if (lastCheckedAt == approxTime.now())
return;
lastCheckedAt = approxTime.now();
if (isAborted())
{
stop();
return true;
throw new QueryCancelledException(ReadCommand.this);
}
return false;
}
private void maybeDelayForTesting()
{
if (!metadata().keyspace.startsWith("system"))
FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
}
}
protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter)
private UnfilteredPartitionIterator withQueryCancellation(UnfilteredPartitionIterator iter)
{
return Transformation.apply(iter, new CheckForAbort());
return Transformation.apply(iter, new QueryCancellationChecker());
}
/**
* A transformation used for simulating slow queries by tests.
*/
@VisibleForTesting
private static class DelayInjector extends Transformation<UnfilteredRowIterator>
{
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
return Transformation.apply(partition, this);
}
@Override
protected Row applyToRow(Row row)
{
FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
return row;
}
}
private UnfilteredPartitionIterator maybeSlowDownForTesting(UnfilteredPartitionIterator iter)
{
if (TEST_ITERATION_DELAY_MILLIS > 0 && !SchemaConstants.isSystemKeyspace(metadata().keyspace))
return Transformation.apply(iter, new DelayInjector());
else
return iter;
}
/**

26
src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java
View File

@ -19,6 +19,8 @@ package org.apache.cassandra.db;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -26,6 +28,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.QueryCancelledException;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
@ -87,17 +90,28 @@ public class ReadCommandVerbHandler implements IVerbHandler<ReadCommand>
{
response = command.createResponse(iterator, controller.getRepairedDataInfo());
}
catch (AssertionError t)
{
throw new AssertionError(String.format("Caught an error while trying to process the command: %s", command.toCQLString()), t);
}
catch (QueryCancelledException e)
{
logger.debug("Query cancelled (timeout)", e);
response = null;
Preconditions.checkState(!command.isCompleted(), "Read marked as completed despite being aborted by timeout to table %s", command.metadata());
}
if (!command.complete())
if (command.complete())
{
Tracing.trace("Enqueuing response to {}", message.from());
Message<ReadResponse> reply = message.responseWith(response);
MessagingService.instance().send(reply, message.from());
}
else
{
Tracing.trace("Discarding partial response to {} (timed out)", message.from());
MessagingService.instance().metrics.recordDroppedMessage(message, message.elapsedSinceCreated(NANOSECONDS), NANOSECONDS);
return;
}
Tracing.trace("Enqueuing response to {}", message.from());
Message<ReadResponse> reply = message.responseWith(response);
MessagingService.instance().send(reply, message.from());
}
private void validateTransientStatus(Message<ReadCommand> message)

50
src/java/org/apache/cassandra/db/transform/RTBoundValidator.java
View File

@ -17,14 +17,16 @@
*/
package org.apache.cassandra.db.transform;
import java.util.Objects;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.schema.TableMetadata;
/**
* A validating transformation that sanity-checks the sequence of RT bounds and boundaries in every partition.
* A validating transformation that sanity-checks the sequence of Range Tombstone bounds and boundaries in every
* partition.
*
* What we validate, specifically:
* - that open markers are only followed by close markers
@ -51,29 +53,27 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
public static UnfilteredRowIterator validate(UnfilteredRowIterator partition, Stage stage, boolean enforceIsClosed)
{
return Transformation.apply(partition, new RowsTransformation(stage, partition.metadata(), partition.isReverseOrder(), enforceIsClosed));
return Transformation.apply(partition, new RowsTransformation(stage, partition, enforceIsClosed));
}
@Override
public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
return Transformation.apply(partition, new RowsTransformation(stage, partition.metadata(), partition.isReverseOrder(), enforceIsClosed));
return Transformation.apply(partition, new RowsTransformation(stage, partition, enforceIsClosed));
}
private final static class RowsTransformation extends Transformation
{
private final Stage stage;
private final TableMetadata metadata;
private final boolean isReverseOrder;
private final boolean enforceIsClosed;
private final UnfilteredRowIterator partition;
private DeletionTime openMarkerDeletionTime;
private RowsTransformation(Stage stage, TableMetadata metadata, boolean isReverseOrder, boolean enforceIsClosed)
private RowsTransformation(Stage stage, UnfilteredRowIterator partition, boolean enforceIsClosed)
{
this.stage = stage;
this.metadata = metadata;
this.isReverseOrder = isReverseOrder;
this.partition = partition;
this.enforceIsClosed = enforceIsClosed;
}
@ -83,25 +83,25 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
if (null == openMarkerDeletionTime)
{
// there is no open RT in the stream - we are expecting a *_START_BOUND
if (marker.isClose(isReverseOrder))
throw ise("unexpected end bound or boundary " + marker.toString(metadata));
if (marker.isClose(partition.isReverseOrder()))
throw ise("unexpected end bound or boundary " + marker.toString(partition.metadata()));
}
else
{
// there is an open RT in the stream - we are expecting a *_BOUNDARY or an *_END_BOUND
if (!marker.isClose(isReverseOrder))
throw ise("start bound followed by another start bound " + marker.toString(metadata));
if (!marker.isClose(partition.isReverseOrder()))
throw ise("start bound followed by another start bound " + marker.toString(partition.metadata()));
// deletion times of open/close markers must match
DeletionTime deletionTime = marker.closeDeletionTime(isReverseOrder);
DeletionTime deletionTime = marker.closeDeletionTime(partition.isReverseOrder());
if (!deletionTime.equals(openMarkerDeletionTime))
throw ise("open marker and close marker have different deletion times");
throw ise("open marker and close marker have different deletion times, close=" + deletionTime);
openMarkerDeletionTime = null;
}
if (marker.isOpen(isReverseOrder))
openMarkerDeletionTime = marker.openDeletionTime(isReverseOrder);
if (marker.isOpen(partition.isReverseOrder()))
openMarkerDeletionTime = marker.openDeletionTime(partition.isReverseOrder());
return marker;
}
@ -110,14 +110,22 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
public void onPartitionClose()
{
if (enforceIsClosed && null != openMarkerDeletionTime)
throw ise("expected all RTs to be closed, but the last one is open");
throw ise("expected all Range Tombstones to be closed, but the last one is open");
}
private IllegalStateException ise(String why)
{
String message =
String.format("%s UnfilteredRowIterator for %s has an illegal RT bounds sequence: %s", stage, metadata, why);
throw new IllegalStateException(message);
throw new IllegalStateException(message(why));
}
private String message(String why)
{
return String.format("%s UnfilteredRowIterator for %s (key: %s omdt: [%s]) has an illegal Range Tombstone bounds sequence: %s",
stage,
partition.metadata(),
partition.metadata().partitionKeyType.getString(partition.partitionKey().getKey()),
Objects.toString(openMarkerDeletionTime, "not present"),
why);
}
}
}

28
src/java/org/apache/cassandra/exceptions/QueryCancelledException.java Normal file
View File

@ -0,0 +1,28 @@
/*
* 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.exceptions;
import org.apache.cassandra.db.ReadCommand;
public class QueryCancelledException extends RuntimeException
{
public QueryCancelledException(ReadCommand command)
{
super("Query cancelled for taking too long: " + command.toCQLString());
}
}

7
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -79,6 +79,7 @@ import org.apache.cassandra.exceptions.CasWriteUnknownResultException;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.IsBootstrappingException;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.exceptions.QueryCancelledException;
import org.apache.cassandra.exceptions.ReadFailureException;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestFailureException;
@ -1986,6 +1987,12 @@ public class StorageProxy implements StorageProxyMBean
{
response = command.createResponse(iterator, controller.getRepairedDataInfo());
}
catch (QueryCancelledException e)
{
logger.debug("Query cancelled (timeout)", e);
response = null;
Preconditions.checkState(!command.isCompleted(), "Local read marked as completed despite being aborted by timeout to table %s", command.metadata());
}
if (command.complete())
{

62
test/distributed/org/apache/cassandra/distributed/test/TimeoutAbortTest.java Normal file
View File

@ -0,0 +1,62 @@
/*
* 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.distributed.test;
import java.io.IOException;
import java.util.List;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class TimeoutAbortTest extends TestBaseImpl
{
@Test
public void timeoutTest() throws IOException, InterruptedException
{
System.setProperty("cassandra.test.read_iteration_delay_ms", "5000");
try (Cluster cluster = init(Cluster.build(1).start()))
{
cluster.schemaChange(withKeyspace("create table %s.tbl (id int, ck1 int, ck2 int, d int, primary key (id, ck1, ck2))"));
cluster.coordinator(1).execute(withKeyspace("delete from %s.tbl using timestamp 5 where id = 1 and ck1 = 77 "), ConsistencyLevel.ALL);
cluster.get(1).flush(KEYSPACE);
Thread.sleep(1000);
for (int i = 0; i < 100; i++)
cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (id, ck1, ck2, d) values (1,77,?,1) using timestamp 10"), ConsistencyLevel.ALL, i);
cluster.get(1).flush(KEYSPACE);
boolean caughtException = false;
try
{
cluster.coordinator(1).execute(withKeyspace("select * from %s.tbl where id=1 and ck1 = 77"), ConsistencyLevel.ALL);
}
catch (Exception e)
{
assertTrue(e.getClass().getName().contains("ReadTimeoutException"));
caughtException = true;
}
assertTrue(caughtException);
List<String> errors = cluster.get(1).logs().grepForErrors().getResult();
assertFalse(errors.toString(), errors.stream().anyMatch(s -> s.contains("open RT bound")));
}
}
}

34
test/unit/org/apache/cassandra/db/ReadCommandTest.java
View File

@ -56,6 +56,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.QueryCancelledException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
@ -229,7 +230,16 @@ public class ReadCommandTest
assertEquals(2, Util.getAll(readCommand).size());
readCommand.abort();
assertEquals(0, Util.getAll(readCommand).size());
boolean cancelled = false;
try
{
Util.getAll(readCommand);
}
catch (QueryCancelledException e)
{
cancelled = true;
}
assertTrue(cancelled);
}
@Test
@ -260,7 +270,16 @@ public class ReadCommandTest
assertEquals(2, partitions.get(0).rowCount());
readCommand.abort();
assertEquals(0, Util.getAll(readCommand).size());
boolean cancelled = false;
try
{
Util.getAll(readCommand);
}
catch (QueryCancelledException e)
{
cancelled = true;
}
assertTrue(cancelled);
}
@Test
@ -291,7 +310,16 @@ public class ReadCommandTest
assertEquals(2, partitions.get(0).rowCount());
readCommand.abort();
assertEquals(0, Util.getAll(readCommand).size());
boolean cancelled = false;
try
{
Util.getAll(readCommand);
}
catch (QueryCancelledException e)
{
cancelled = true;
}
assertTrue(cancelled);
}
@Test