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

Fix upgrade TTL tests that fail due to improved error detection in cep-15-accord 

patch by David Capwell; reviewed by Caleb Rackliffe for CASSANDRA-20535
This commit is contained in:
David Capwell 2025-04-07 09:08:49 -07:00
parent 265511f6de
commit 92212b2255
11 changed files with 90 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/cql3/selection/ColumnTimestamps.java
View File

@ -27,7 +27,6 @@ import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UserType;
@ -385,7 +384,7 @@ abstract class ColumnTimestamps
List<ByteBuffer> buffers = new ArrayList<>(timestamps.size());
timestamps.forEach(timestamp -> buffers.add(type.toByteBuffer(timestamp)));
return LONG_LIST_TYPE.pack(buffers, ByteBufferAccessor.instance);
return LONG_LIST_TYPE.pack(buffers);
}
@Override

3
src/java/org/apache/cassandra/cql3/selection/ListSelector.java
View File

@ -29,7 +29,6 @@ import org.apache.cassandra.cql3.terms.Lists;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.filter.ColumnFilter.Builder;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -102,7 +101,7 @@ final class ListSelector extends Selector
{
buffers.add(elements.get(i).getOutput(protocolVersion));
}
return type.pack(buffers, ByteBufferAccessor.instance);
return type.pack(buffers);
}
public void reset()

3
src/java/org/apache/cassandra/cql3/selection/MapSelector.java
View File

@ -34,7 +34,6 @@ import org.apache.cassandra.cql3.functions.Function;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.filter.ColumnFilter.Builder;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -218,7 +217,7 @@ final class MapSelector extends Selector
buffers.add(entry.getKey());
buffers.add(entry.getValue());
}
return type.pack(buffers, ByteBufferAccessor.instance);
return type.pack(buffers);
}
public void reset()

3
src/java/org/apache/cassandra/cql3/selection/SetSelector.java
View File

@ -31,7 +31,6 @@ import org.apache.cassandra.cql3.terms.Sets;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.filter.ColumnFilter.Builder;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -104,7 +103,7 @@ final class SetSelector extends Selector
{
buffers.add(elements.get(i).getOutput(protocolVersion));
}
return type.pack(new ArrayList<>(buffers), ByteBufferAccessor.instance);
return type.pack(new ArrayList<>(buffers));
}
public void reset()

3
src/java/org/apache/cassandra/cql3/selection/TupleSelector.java
View File

@ -29,7 +29,6 @@ import org.apache.cassandra.cql3.terms.Tuples;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.filter.ColumnFilter.Builder;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.TupleType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.io.util.DataInputPlus;
@ -103,7 +102,7 @@ final class TupleSelector extends Selector
{
buffers.add(elements.get(i).getOutput(protocolVersion));
}
return type.pack(buffers, ByteBufferAccessor.instance);
return type.pack(buffers);
}
public void reset()

3
src/java/org/apache/cassandra/cql3/selection/UserTypeSelector.java
View File

@ -27,7 +27,6 @@ import java.util.Map.Entry;
import com.google.common.base.Objects;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.ColumnSpecification;
@ -198,7 +197,7 @@ final class UserTypeSelector extends Selector
Selector selector = fields.get(userType.fieldName(i));
buffers.add(selector == null ? null : selector.getOutput(protocolVersion));
}
return type.pack(buffers, ByteBufferAccessor.instance);
return type.pack(buffers);
}
public void reset()

3
src/java/org/apache/cassandra/cql3/selection/VectorSelector.java
View File

@ -30,7 +30,6 @@ import org.apache.cassandra.cql3.terms.Lists;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.VectorType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.io.util.DataInputPlus;
@ -127,7 +126,7 @@ public class VectorSelector extends Selector
for (int i = 0, m = elements.size(); i < m; i++)
buffers.add(elements.get(i).getOutput(protocolVersion));
return type.pack(buffers, ByteBufferAccessor.instance);
return type.pack(buffers);
}
@Override

2
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -1154,7 +1154,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
.collect(Collectors.toList());
try
{
FBUtilities.waitOnFutures(futures, instances.size(), TimeUnit.MINUTES);
FBUtilities.waitOnFutures(futures, 1L, TimeUnit.MINUTES);
}
catch (Throwable t)
{

2
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeBatchTestBase.java
View File

@ -62,6 +62,8 @@ public class MixedModeBatchTestBase extends UpgradeTestBase
.setup(cluster -> {
cluster.schemaChange("CREATE KEYSPACE test_simple WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};");
cluster.schemaChange("CREATE TABLE test_simple.names (key int PRIMARY KEY, name text)");
if (isLogged)
cluster.setUncaughtExceptionsFilter(t -> t.getMessage() != null && t.getMessage().startsWith("Operation timed out"));
})
.runAfterNodeUpgrade((cluster, upgraded) -> {
if (isLogged)

36
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeTTLOverflowUpgradeTestBase.java
View File

@ -22,6 +22,7 @@ import java.util.List;
import java.util.function.BiConsumer;
import java.util.stream.Stream;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.cql3.Attributes;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.Feature;
@ -77,8 +78,35 @@ public abstract class MixedModeTTLOverflowUpgradeTestBase extends UpgradeTestBas
static volatile long clusterStatupTime = 0;
private static class AllowedErrors
{
private final IntHashSet upgraded = new IntHashSet();
private void clear()
{
upgraded.clear();
}
private void upgraded(int node)
{
upgraded.add(node);
}
private boolean uncaughtExceptionsFilter(int node, Throwable t)
{
String message = t.getMessage();
if (message != null && message.endsWith("In order to avoid this use a lower TTL, change the expiration date overflow policy or upgrade to a version where this limitation is fixed. See CASSANDRA-14092 for more details."))
{
// upgraded nodes should not produce these errors
return !upgraded.contains(node);
}
return false;
}
}
static void testTTLOverflow(RunOnClusterAndNode runAfterNodeUpgrade) throws Throwable
{
AllowedErrors allowedErrors = new AllowedErrors();
new TestCase()
.nodes(2)
.nodesToUpgradeOrdered(1, 2)
@ -87,6 +115,9 @@ public abstract class MixedModeTTLOverflowUpgradeTestBase extends UpgradeTestBas
.singleUpgradeToCurrentFrom(v41)
.withConfig(c -> c.with(Feature.GOSSIP).set("storage_compatibility_mode", "CASSANDRA_4"))
.setup(cluster -> {
allowedErrors.clear();
cluster.setUncaughtExceptionsFilter(allowedErrors::uncaughtExceptionsFilter);
cluster.schemaChange(String.format("CREATE TABLE %s.%s (k int PRIMARY KEY, v1 int, v2 int)", KEYSPACE, T_REGULAR));
cluster.schemaChange(String.format("CREATE TABLE %s.%s (k int, c int, v1 int, v2 int, PRIMARY KEY (k, c))", KEYSPACE, T_CLUST));
cluster.schemaChange(String.format("CREATE TABLE %s.%s (k int, c int, v1 int static, v2 int, PRIMARY KEY (k, c))", KEYSPACE, T_STATIC));
@ -100,7 +131,10 @@ public abstract class MixedModeTTLOverflowUpgradeTestBase extends UpgradeTestBas
clusterStatupTime = Clock.Global.currentTimeMillis();
verify(Step.NODE1_PREV_NODE2_PREV, cluster, true);
})
.runAfterNodeUpgrade(runAfterNodeUpgrade)
.runAfterNodeUpgrade((c, n) -> {
allowedErrors.upgraded(n);
runAfterNodeUpgrade.run(c, n);
})
.run();
}

50
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeWritetimeOrTTLTest.java
View File

@ -20,9 +20,11 @@ package org.apache.cassandra.distributed.upgrade;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import org.junit.Test;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.assertj.core.api.Assertions;
@ -40,9 +42,14 @@ import static org.apache.cassandra.distributed.shared.AssertUtils.row;
*/
public class MixedModeWritetimeOrTTLTest extends UpgradeTestBase
{
private static final String CANNOT_USE_SELECTION_FUNCTION_WRITE_TIME_ON_NON_FROZEN_COLLECTION_S = "Cannot use selection function writeTime on non-frozen collection s";
private static final String CANNOT_USE_SELECTION_FUNCTION_TTL_ON_NON_FROZEN_COLLECTION_S = "Cannot use selection function ttl on non-frozen collection s";
private static final String MAXWRITETIME_UNKNOWN = "Unknown function 'maxwritetime'";
@Test
public void testWritetimeOrTTLDuringUpgrade() throws Throwable
{
AllowedErrors allowedErrors = new AllowedErrors();
new TestCase()
.nodes(2)
.nodesToUpgradeOrdered(1, 2)
@ -51,6 +58,8 @@ public class MixedModeWritetimeOrTTLTest extends UpgradeTestBase
.singleUpgradeToCurrentFrom(v41)
.withConfig(c -> c.with(Feature.GOSSIP))
.setup(cluster -> {
allowedErrors.clear();
cluster.setUncaughtExceptionsFilter(allowedErrors::uncaughtExceptionsFilter);
ICoordinator coordinator = cluster.coordinator(1);
cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k int PRIMARY KEY, v int, s set<int>, fs frozen<set<int>>)"));
@ -61,6 +70,7 @@ public class MixedModeWritetimeOrTTLTest extends UpgradeTestBase
assertPre42Behaviour(cluster.coordinator(2));
})
.runAfterNodeUpgrade((cluster, node) -> {
allowedErrors.upgraded(node);
if (node == 1) // only node1 is upgraded, and the cluster is in mixed mode
{
assertPost42Behaviour(cluster.coordinator(1));
@ -80,24 +90,24 @@ public class MixedModeWritetimeOrTTLTest extends UpgradeTestBase
// regular column, supported except for maxwritetime
assertRows(coordinator.execute(withKeyspace("SELECT writetime(v) FROM %s.t"), ALL), row(2L));
Assertions.assertThatThrownBy(() -> coordinator.execute(withKeyspace("SELECT maxwritetime(v) FROM %s.t"), ALL))
.hasMessageContaining("Unknown function 'maxwritetime'");
.hasMessageContaining(MAXWRITETIME_UNKNOWN);
Assertions.assertThat((Integer) coordinator.execute(withKeyspace("SELECT ttl(v) FROM %s.t"), ALL)[0][0])
.isLessThanOrEqualTo(2000).isGreaterThan(2000 - 300); // margin of error of 5 minutes since TTLs decrease
// frozen collection, supported except for maxwritetime
assertRows(coordinator.execute(withKeyspace("SELECT writetime(fs) FROM %s.t"), ALL), row(1L));
Assertions.assertThatThrownBy(() -> coordinator.execute(withKeyspace("SELECT maxwritetime(fs) FROM %s.t"), ALL))
.hasMessageContaining("Unknown function 'maxwritetime'");
.hasMessageContaining(MAXWRITETIME_UNKNOWN);
Assertions.assertThat((Integer) coordinator.execute(withKeyspace("SELECT ttl(fs) FROM %s.t"), ALL)[0][0])
.isLessThanOrEqualTo(1000).isGreaterThan(1000 - 300); // margin of error of 5 minutes since TTLs decrease
// not-frozen collection, not supported
Assertions.assertThatThrownBy(() -> coordinator.execute(withKeyspace("SELECT writetime(s) FROM %s.t"), ALL))
.hasMessageContaining("Cannot use selection function writeTime on non-frozen collection s");
.hasMessageContaining(CANNOT_USE_SELECTION_FUNCTION_WRITE_TIME_ON_NON_FROZEN_COLLECTION_S);
Assertions.assertThatThrownBy(() -> coordinator.execute(withKeyspace("SELECT maxwritetime(s) FROM %s.t"), ALL))
.hasMessageContaining("Unknown function 'maxwritetime'");
.hasMessageContaining(MAXWRITETIME_UNKNOWN);
Assertions.assertThatThrownBy(() -> coordinator.execute(withKeyspace("SELECT ttl(s) FROM %s.t"), ALL))
.hasMessageContaining("Cannot use selection function ttl on non-frozen collection s");
.hasMessageContaining(CANNOT_USE_SELECTION_FUNCTION_TTL_ON_NON_FROZEN_COLLECTION_S);
}
private void assertPost42Behaviour(ICoordinator coordinator)
@ -120,4 +130,34 @@ public class MixedModeWritetimeOrTTLTest extends UpgradeTestBase
Assertions.assertThat(coordinator.execute(withKeyspace("SELECT ttl(s) FROM %s.t"), ALL)[0][0])
.matches(l -> l instanceof List && ((List<?>) l).size() == 4);
}
private static class AllowedErrors
{
private static final Set<String> EXPECTED_ERRORS = Set.of(MAXWRITETIME_UNKNOWN,
CANNOT_USE_SELECTION_FUNCTION_WRITE_TIME_ON_NON_FROZEN_COLLECTION_S,
CANNOT_USE_SELECTION_FUNCTION_TTL_ON_NON_FROZEN_COLLECTION_S);
private final IntHashSet upgraded = new IntHashSet();
private void clear()
{
upgraded.clear();
}
private void upgraded(int node)
{
upgraded.add(node);
}
private boolean uncaughtExceptionsFilter(int node, Throwable t)
{
String message = t.getMessage();
if (message != null && EXPECTED_ERRORS.contains(message))
{
// upgraded nodes should not produce these errors
return !upgraded.contains(node);
}
return false;
}
}
}