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

Fix a race condition where a keyspace can be opened while it is being removed 

patch by Jacek Lewandowski; reviewed by Andrés de la Peña and Ekaterina Dimitrova for CASSANDRA 17658
This commit is contained in:
Jacek Lewandowski 2022-06-10 11:43:53 +02:00
parent ab3b59ec42
commit 570732375e
20 changed files with 336 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -28,6 +28,7 @@
* Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495)
* Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)
Merged from 4.1:
* Fix a race condition where a keyspace can be oopened while it is being removed (CASSANDRA-17658)
* DatabaseDescriptor will set the default failure detector during client initialization (CASSANDRA-17782)
* Avoid initializing schema via SystemKeyspace.getPreferredIP() with the BulkLoader tool (CASSANDRA-17740)
* Uncomment prepared_statements_cache_size, key_cache_size, counter_cache_size, index_summary_capacity which were

10
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -126,7 +126,10 @@ public class Keyspace
public static void setInitialized()
{
initialized = true;
synchronized (Schema.instance)
{
initialized = true;
}
}
/**
@ -137,7 +140,10 @@ public class Keyspace
@VisibleForTesting
public static void unsetInitialized()
{
initialized = false;
synchronized (Schema.instance)
{
initialized = false;
}
}
public static Keyspace open(String keyspaceName)

13
src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java
View File

@ -24,7 +24,6 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
@ -47,7 +46,6 @@ import org.apache.cassandra.schema.SchemaTransformation.SchemaTransformationResu
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import static org.apache.cassandra.schema.MigrationCoordinator.MAX_OUTSTANDING_VERSION_REQUESTS;
@ -257,12 +255,11 @@ public class DefaultSchemaUpdateHandler implements SchemaUpdateHandler, IEndpoin
@Override
public SchemaTransformationResult reset(boolean local)
{
return local
? reload()
: migrationCoordinator.pullSchemaFromAnyNode()
.flatMap(mutations -> ImmediateFuture.success(applyMutations(mutations)))
.awaitThrowUncheckedOnInterrupt()
.getNow();
if (local)
return reload();
Collection<Mutation> mutations = migrationCoordinator.pullSchemaFromAnyNode().awaitThrowUncheckedOnInterrupt().getNow();
return applyMutations(mutations);
}
@Override

58
src/java/org/apache/cassandra/schema/Schema.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.schema;
import java.time.Duration;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
@ -227,15 +228,15 @@ public class Schema implements SchemaProvider
return keyspaceInstances.blockingLoadIfAbsent(keyspaceName, loadFunction);
}
public Keyspace maybeRemoveKeyspaceInstance(String keyspaceName, boolean dropData)
private Keyspace maybeRemoveKeyspaceInstance(String keyspaceName, Consumer<Keyspace> unloadFunction)
{
try
{
return keyspaceInstances.blockingUnloadIfPresent(keyspaceName, keyspace -> keyspace.unload(dropData));
return keyspaceInstances.blockingUnloadIfPresent(keyspaceName, unloadFunction);
}
catch (LoadingMap.UnloadExecutionException e)
{
throw new AssertionError("Failed to unload the keyspace " + keyspaceName);
throw new AssertionError("Failed to unload the keyspace " + keyspaceName, e);
}
}
@ -532,16 +533,6 @@ public class Schema implements SchemaProvider
SchemaDiagnostics.versionUpdated(this);
}
/**
* Clear all KS/CF metadata and reset version.
*/
public synchronized void clear()
{
distributedKeyspaces.forEach(this::unload);
updateVersion(SchemaConstants.emptyVersion);
SchemaDiagnostics.schemaCleared(this);
}
/**
* When we receive {@link SchemaTransformationResult} in a callback invocation, the transformation result includes
* pre-transformation and post-transformation schema metadata and versions, and a diff between them. Basically
@ -617,16 +608,18 @@ public class Schema implements SchemaProvider
}
/**
* Clear all locally stored schema information and reset schema to initial state.
* Clear all locally stored schema information and fetch schema from another node.
* Called by user (via JMX) who wants to get rid of schema disagreement.
*/
public void resetLocalSchema()
public synchronized void resetLocalSchema()
{
logger.debug("Clearing local schema...");
updateHandler.clear();
logger.debug("Clearing local schema keyspace instances...");
clear();
distributedKeyspaces.forEach(this::unload);
updateVersion(SchemaConstants.emptyVersion);
SchemaDiagnostics.schemaCleared(this);
updateHandler.reset(false);
logger.info("Local schema reset is complete.");
@ -692,37 +685,40 @@ public class Schema implements SchemaProvider
// we send mutations to the correct set of bootstrapping nodes. Refer CASSANDRA-15433.
if (keyspace.params.replication.klass != LocalStrategy.class && Keyspace.isInitialized())
{
PendingRangeCalculatorService.calculatePendingRanges(Keyspace.open(keyspace.name).getReplicationStrategy(), keyspace.name);
PendingRangeCalculatorService.calculatePendingRanges(Keyspace.open(keyspace.name, this, true).getReplicationStrategy(), keyspace.name);
}
}
private void dropKeyspace(KeyspaceMetadata keyspace, boolean dropData)
private void dropKeyspace(KeyspaceMetadata keyspaceMetadata, boolean dropData)
{
SchemaDiagnostics.keyspaceDropping(this, keyspace);
SchemaDiagnostics.keyspaceDropping(this, keyspaceMetadata);
boolean initialized = Keyspace.isInitialized();
Keyspace ks = initialized ? getKeyspaceInstance(keyspace.name) : null;
Keyspace keyspace = initialized ? Keyspace.open(keyspaceMetadata.name, this, false) : null;
if (initialized)
{
if (ks == null)
if (keyspace == null)
return;
keyspace.views.forEach(v -> dropView(ks, v, dropData));
keyspace.tables.forEach(t -> dropTable(ks, t, dropData));
keyspaceMetadata.views.forEach(v -> dropView(keyspace, v, dropData));
keyspaceMetadata.tables.forEach(t -> dropTable(keyspace, t, dropData));
// remove the keyspace from the static instances
maybeRemoveKeyspaceInstance(keyspace.name, dropData);
}
Keyspace unloadedKeyspace = maybeRemoveKeyspaceInstance(keyspaceMetadata.name, ks -> {
ks.unload(dropData);
unload(keyspaceMetadata);
});
assert unloadedKeyspace == keyspace;
unload(keyspace);
if (initialized)
{
Keyspace.writeOrder.awaitNewBarrier();
}
else
{
unload(keyspaceMetadata);
}
schemaChangeNotifier.notifyKeyspaceDropped(keyspace, dropData);
SchemaDiagnostics.keyspaceDropped(this, keyspace);
schemaChangeNotifier.notifyKeyspaceDropped(keyspaceMetadata, dropData);
SchemaDiagnostics.keyspaceDropped(this, keyspaceMetadata);
}
private void dropView(Keyspace keyspace, ViewMetadata metadata, boolean dropData)

34
test/distributed/org/apache/cassandra/distributed/test/SchemaTest.java
View File

@ -18,10 +18,15 @@
package org.apache.cassandra.distributed.test;
import java.time.Duration;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.schema.Schema;
import org.awaitility.Awaitility;
import static org.junit.Assert.assertTrue;
@ -86,4 +91,33 @@ public class SchemaTest extends TestBaseImpl
assertTrue(causeIsUnknownColumn);
}
}
@Test
public void schemaReset() throws Throwable
{
try (Cluster cluster = init(Cluster.build(2).withConfig(cfg -> cfg.with(Feature.GOSSIP, Feature.NETWORK)).start()))
{
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk INT PRIMARY KEY, v TEXT)");
assertTrue(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null));
assertTrue(cluster.get(2).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null));
cluster.get(2).shutdown().get();
// when schema is removed and there is no other node to fetch it from, node 1 should be left with clean schema
cluster.get(1).runOnInstance(() -> Schema.instance.resetLocalSchema());
assertTrue(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") == null));
// when the other node is started, schema should be back in sync
cluster.get(2).startup();
Awaitility.waitAtMost(Duration.ofMinutes(1))
.pollDelay(Duration.ofSeconds(1))
.until(() -> cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null));
// when schema is removed and there is a node to fetch it from, node 1 should immediatelly restore the schema
cluster.get(1).runOnInstance(() -> Schema.instance.resetLocalSchema());
assertTrue(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null));
}
}
}

21
test/long/org/apache/cassandra/cql3/CorruptionTest.java
View File

@ -26,7 +26,10 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.io.util.File;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@ -34,14 +37,12 @@ import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.LoggingRetryPolicy;
import com.datastax.driver.core.policies.Policies;
import com.datastax.driver.core.utils.Bytes;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.FileWriter;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.EmbeddedCassandraService;
public class CorruptionTest extends SchemaLoader
public class CorruptionTest
{
private static EmbeddedCassandraService cassandra;
@ -59,10 +60,7 @@ public class CorruptionTest extends SchemaLoader
@BeforeClass()
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
cassandra = new EmbeddedCassandraService();
cassandra.start();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = Cluster.builder().addContactPoint("127.0.0.1")
.withRetryPolicy(new LoggingRetryPolicy(Policies.defaultRetryPolicy()))
@ -102,6 +100,15 @@ public class CorruptionTest extends SchemaLoader
VALUE = s.toString();
}
@AfterClass
public static void tearDown()
{
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
@Test
public void runCorruptionTest()
{

11
test/unit/org/apache/cassandra/ServerTestUtils.java
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.locator.AbstractEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.service.EmbeddedCassandraService;
/**
* Utility methodes used by SchemaLoader and CQLTester to manage the server and its state.
@ -189,6 +190,16 @@ public final class ServerTestUtils
cleanupDirectory(DatabaseDescriptor.getSavedCachesLocation());
}
public static EmbeddedCassandraService startEmbeddedCassandraService() throws IOException
{
DatabaseDescriptor.daemonInitialization();
mkdirs();
cleanup();
EmbeddedCassandraService service = new EmbeddedCassandraService();
service.start();
return service;
}
private ServerTestUtils()
{
}

7
test/unit/org/apache/cassandra/audit/AuditLoggerAuthTest.java
View File

@ -33,11 +33,10 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.AuthenticationException;
import com.datastax.driver.core.exceptions.SyntaxError;
import com.datastax.driver.core.exceptions.UnauthorizedException;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.OverrideConfigurationLoader;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.PasswordObfuscator;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.EmbeddedCassandraService;
@ -75,11 +74,9 @@ public class AuditLoggerAuthTest
config.audit_logging_options.enabled = true;
config.audit_logging_options.logger = new ParameterizedClass("InMemoryAuditLogger", null);
});
CQLTester.prepareServer();
System.setProperty("cassandra.superuser_setup_delay_ms", "0");
embedded = new EmbeddedCassandraService();
embedded.start();
embedded = ServerTestUtils.startEmbeddedCassandraService();
executeWithCredentials(
Arrays.asList(getCreateRoleCql(TEST_USER, true, false, false),

15
test/unit/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutorTest.java
View File

@ -26,10 +26,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.Assert;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.apache.cassandra.service.StorageService;
@ -43,10 +46,14 @@ public class DebuggableScheduledThreadPoolExecutorTest
@BeforeClass
public static void startup() throws IOException
{
//The DSTPE checks for if we are in the service shutdown hook so
//to test it we need to start C* internally.
service = new EmbeddedCassandraService();
service.start();
service = ServerTestUtils.startEmbeddedCassandraService();
}
@AfterClass
public static void tearDown()
{
if (service != null)
service.stop();
}
@Test

17
test/unit/org/apache/cassandra/cql3/BatchTests.java
View File

@ -22,15 +22,18 @@ import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.io.IOException;
public class BatchTests extends CQLTester
public class BatchTests
{
private static EmbeddedCassandraService cassandra;
@ -44,8 +47,7 @@ public class BatchTests extends CQLTester
@BeforeClass()
public static void setup() throws ConfigurationException, IOException
{
cassandra = new EmbeddedCassandraService();
cassandra.start();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
@ -75,6 +77,15 @@ public class BatchTests extends CQLTester
clustering = session.prepare("insert into junit.clustering(id, clustering1, clustering2, clustering3, val) values(?,?,?,?,?)");
}
@AfterClass
public static void tearDown()
{
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
@Test(expected = InvalidQueryException.class)
public void testMixedInCounterBatch()
{

6
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -429,12 +429,14 @@ public abstract class CQLTester
public static List<String> buildNodetoolArgs(List<String> args)
{
int port = jmxPort == 0 ? Integer.getInteger("cassandra.jmx.local.port", 7199) : jmxPort;
String host = jmxHost == null ? "127.0.0.1" : jmxHost;
List<String> allArgs = new ArrayList<>();
allArgs.add("bin/nodetool");
allArgs.add("-p");
allArgs.add(Integer.toString(jmxPort));
allArgs.add(String.valueOf(port));
allArgs.add("-h");
allArgs.add(jmxHost == null ? "127.0.0.1" : jmxHost);
allArgs.add(host);
allArgs.addAll(args);
return allArgs;
}

11
test/unit/org/apache/cassandra/cql3/PagingTest.java
View File

@ -29,6 +29,7 @@ import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
@ -51,13 +52,14 @@ public class PagingTest
" WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 2 };";
private static final String dropKsStatement = "DROP KEYSPACE IF EXISTS " + KEYSPACE;
private static EmbeddedCassandraService cassandra;
@BeforeClass
public static void setup() throws Exception
{
System.setProperty("cassandra.config", "cassandra-murmur.yaml");
EmbeddedCassandraService cassandra = new EmbeddedCassandraService();
cassandra.start();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
// Currently the native server start method return before the server is fully binded to the socket, so we need
// to wait slightly before trying to connect to it. We should fix this but in the meantime using a sleep.
@ -75,7 +77,10 @@ public class PagingTest
@AfterClass
public static void tearDown()
{
cluster.close();
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
/**

26
test/unit/org/apache/cassandra/metrics/BatchMetricsTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.metrics;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@ -28,20 +29,20 @@ import com.datastax.driver.core.BatchStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.EmbeddedCassandraService;
import static org.apache.cassandra.cql3.statements.BatchStatement.metrics;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.*;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.EstimatedHistogramReservoirSnapshot;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.Range;
import static org.junit.Assert.assertEquals;
import static org.quicktheories.QuickTheory.qt;
import static org.quicktheories.generators.Generate.intArrays;
import static org.quicktheories.generators.SourceDSL.integers;
public class BatchMetricsTest extends SchemaLoader
public class BatchMetricsTest
{
private static final int MAX_ROUNDS_TO_PERFORM = 3;
private static final int MAX_DISTINCT_PARTITIONS = 128;
@ -62,13 +63,11 @@ public class BatchMetricsTest extends SchemaLoader
@BeforeClass()
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
cassandra = new EmbeddedCassandraService();
cassandra.start();
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.setWriteRpcTimeout(TimeUnit.SECONDS.toMillis(10));
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
@ -81,6 +80,15 @@ public class BatchMetricsTest extends SchemaLoader
psCounter = session.prepare("UPDATE " + KEYSPACE + '.' + COUNTER_TABLE + " SET val = val + 1 WHERE id = ?;");
}
@AfterClass
public static void tearDown()
{
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
private void executeLoggerBatch(BatchStatement.Type batchStatementType, int distinctPartitions, int statementsPerPartition)
{
BatchStatement batch = new BatchStatement(batchStatementType);

21
test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.metrics;
import java.io.IOException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
@ -28,9 +29,8 @@ import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.EmbeddedCassandraService;
@ -39,18 +39,16 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class CQLMetricsTest extends SchemaLoader
public class CQLMetricsTest
{
private static Cluster cluster;
private static Session session;
private static EmbeddedCassandraService cassandra;
@BeforeClass()
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
EmbeddedCassandraService cassandra = new EmbeddedCassandraService();
cassandra.start();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
@ -59,6 +57,15 @@ public class CQLMetricsTest extends SchemaLoader
session.execute("CREATE TABLE IF NOT EXISTS junit.metricstest (id int PRIMARY KEY, val text);");
}
@AfterClass
public static void tearDown()
{
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
@Test
public void testConnectionWithUseDisabled()
{

23
test/unit/org/apache/cassandra/metrics/ClientRequestMetricsTest.java
View File

@ -29,16 +29,15 @@ import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.EmbeddedCassandraService;
import static com.datastax.driver.core.Cluster.*;
import static com.datastax.driver.core.Cluster.builder;
import static org.junit.Assert.assertEquals;
public class ClientRequestMetricsTest extends SchemaLoader
public class ClientRequestMetricsTest
{
private static Cluster cluster;
private static Session session;
@ -54,13 +53,12 @@ public class ClientRequestMetricsTest extends SchemaLoader
private static final ClientRequestMetrics readMetrics = ClientRequestsMetricsHolder.readMetrics;
private static final ClientWriteRequestMetrics writeMetrics = ClientRequestsMetricsHolder.writeMetrics;
private static EmbeddedCassandraService cassandra;
@BeforeClass
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
EmbeddedCassandraService cassandra = new EmbeddedCassandraService();
cassandra.start();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
@ -74,11 +72,14 @@ public class ClientRequestMetricsTest extends SchemaLoader
readPS = session.prepare("SELECT * FROM " + KEYSPACE + '.' + TABLE + " WHERE id=?;");
readRangePS = session.prepare("SELECT * FROM " + KEYSPACE + '.' + TABLE + " WHERE id=? AND ord>=? AND ord <= ?;");
}
@AfterClass
public static void teardown()
public static void tearDown()
{
cluster.close();
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
@Test

36
test/unit/org/apache/cassandra/metrics/KeyspaceMetricsTest.java
View File

@ -18,40 +18,37 @@
package org.apache.cassandra.metrics;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.EmbeddedCassandraService;
public class KeyspaceMetricsTest extends SchemaLoader
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class KeyspaceMetricsTest
{
private static Session session;
private static Cluster cluster;
private static EmbeddedCassandraService cassandra;
@BeforeClass
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
EmbeddedCassandraService cassandra = new EmbeddedCassandraService();
cassandra.start();
Cluster cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
}
@ -73,10 +70,13 @@ public class KeyspaceMetricsTest extends SchemaLoader
// no metrics after drop
assertEquals(metrics.get().collect(Collectors.joining(",")), 0, metrics.get().count());
}
@AfterClass
public static void teardown()
public static void tearDown()
{
session.close();
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
}

23
test/unit/org/apache/cassandra/metrics/TableMetricsTest.java
View File

@ -31,17 +31,16 @@ import com.datastax.driver.core.BatchStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.EmbeddedCassandraService;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class TableMetricsTest extends SchemaLoader
public class TableMetricsTest
{
private static Session session;
@ -49,15 +48,15 @@ public class TableMetricsTest extends SchemaLoader
private static final String TABLE = "tablemetricstest";
private static final String COUNTER_TABLE = "tablemetricscountertest";
private static EmbeddedCassandraService cassandra;
private static Cluster cluster;
@BeforeClass
public static void setup() throws ConfigurationException, IOException
{
Schema.instance.clear();
cassandra = ServerTestUtils.startEmbeddedCassandraService();
EmbeddedCassandraService cassandra = new EmbeddedCassandraService();
cassandra.start();
Cluster cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
session = cluster.connect();
session.execute(String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };", KEYSPACE));
@ -276,9 +275,13 @@ public class TableMetricsTest extends SchemaLoader
assertEquals(metrics.get().collect(Collectors.joining(",")), 0, metrics.get().count());
}
@AfterClass
public static void teardown()
public static void tearDown()
{
session.close();
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
}

41
test/unit/org/apache/cassandra/schema/SchemaTest.java
View File

@ -25,37 +25,33 @@ import java.util.Collection;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.utils.FBUtilities;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
public class SchemaTest
{
@BeforeClass
public static void setupDatabaseDescriptor()
public static void setup()
{
DatabaseDescriptor.daemonInitialization();
ServerTestUtils.prepareServer();
Schema.instance.loadFromDisk();
}
@Test
public void testTransKsMigration() throws IOException
{
CommitLog.instance.start();
SchemaLoader.cleanupAndLeaveDirs();
Schema.instance.loadFromDisk();
assertEquals(0, Schema.instance.getNonSystemKeyspaces().size());
Gossiper.instance.start((int) (System.currentTimeMillis() / 1000));
Keyspace.setInitialized();
try
{
// add a few.
@ -82,6 +78,33 @@ public class SchemaTest
}
}
@Test
public void testKeyspaceCreationWhenNotInitialized() {
Keyspace.unsetInitialized();
try
{
SchemaTestUtil.addOrUpdateKeyspace(KeyspaceMetadata.create("test", KeyspaceParams.simple(1)), true);
assertNotNull(Schema.instance.getKeyspaceMetadata("test"));
assertNull(Schema.instance.getKeyspaceInstance("test"));
SchemaTestUtil.dropKeyspaceIfExist("test", true);
assertNull(Schema.instance.getKeyspaceMetadata("test"));
assertNull(Schema.instance.getKeyspaceInstance("test"));
}
finally
{
Keyspace.setInitialized();
}
SchemaTestUtil.addOrUpdateKeyspace(KeyspaceMetadata.create("test", KeyspaceParams.simple(1)), true);
assertNotNull(Schema.instance.getKeyspaceMetadata("test"));
assertNotNull(Schema.instance.getKeyspaceInstance("test"));
SchemaTestUtil.dropKeyspaceIfExist("test", true);
assertNull(Schema.instance.getKeyspaceMetadata("test"));
assertNull(Schema.instance.getKeyspaceInstance("test"));
}
private void saveKeyspaces()
{
Collection<Mutation> mutations = Arrays.asList(SchemaKeyspace.makeCreateKeyspaceMutation(KeyspaceMetadata.create("ks0", KeyspaceParams.simple(3)), FBUtilities.timestampMicros()).build(),

147
test/unit/org/apache/cassandra/tools/nodetool/ClientStatsTest.java
View File

@ -18,95 +18,124 @@
package org.apache.cassandra.tools.nodetool;
import java.net.InetAddress;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.LoggerFactory;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.service.CassandraDaemon;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tools.ToolRunner;
import static org.assertj.core.api.Assertions.assertThat;
import org.assertj.core.groups.Tuple;
public class ClientStatsTest extends CQLTester
import static org.assertj.core.api.Assertions.assertThat;
public class ClientStatsTest
{
private static Cluster cluster;
private Session session;
private static EmbeddedCassandraService cassandra;
@BeforeClass
public static void setup() throws Throwable
{
CassandraDaemon daemon = new CassandraDaemon();
requireNetwork();
startJMXServer();
daemon.activate();
daemon.startNativeTransport();
StorageService.instance.registerDaemon(daemon);
// Since we run EmbeddedCassandraServer, we need to manually associate JMX address; otherwise it won't start
int jmxPort = CQLTester.getAutomaticallyAllocatedPort(InetAddress.getLoopbackAddress());
System.setProperty("cassandra.jmx.local.port", String.valueOf(jmxPort));
cassandra = ServerTestUtils.startEmbeddedCassandraService();
cluster = Cluster.builder().addContactPoint("127.0.0.1").withPort(DatabaseDescriptor.getNativeTransportPort()).build();
}
@Before
public void config() throws Throwable
{
ResultSet result = executeNet("select release_version from system.local");
session = cluster.connect();
ResultSet result = session.execute("select release_version from system.local");
}
@After
public void afterTest()
{
if (session != null)
session.close();
}
@AfterClass
public static void tearDown()
{
if (cluster != null)
cluster.close();
if (cassandra != null)
cassandra.stop();
}
@Test
public void testClientStatsHelp()
{
ToolRunner.ToolResult tool = ToolRunner.invokeNodetool("help", "clientstats");
tool.assertOnCleanExit();
String help = "NAME\n" +
" nodetool clientstats - Print information about connected clients\n" +
"\n" +
"SYNOPSIS\n" +
" nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]\n" +
" [(-pp | --print-port)] [(-pw <password> | --password <password>)]\n" +
" [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]\n" +
" [(-u <username> | --username <username>)] clientstats [--all]\n" +
String help = "NAME\n" +
" nodetool clientstats - Print information about connected clients\n" +
"\n" +
"SYNOPSIS\n" +
" nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]\n" +
" [(-pp | --print-port)] [(-pw <password> | --password <password>)]\n" +
" [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]\n" +
" [(-u <username> | --username <username>)] clientstats [--all]\n" +
" [--by-protocol] [--clear-history] [--client-options]\n" +
"\n" +
"OPTIONS\n" +
" --all\n" +
" Lists all connections\n" +
"\n" +
" --by-protocol\n" +
" Lists most recent client connections by protocol version\n" +
"\n" +
" --clear-history\n" +
" Clear the history of connected clients\n" +
"\n" +
"OPTIONS\n" +
" --all\n" +
" Lists all connections\n" +
"\n" +
" --by-protocol\n" +
" Lists most recent client connections by protocol version\n" +
"\n" +
" --clear-history\n" +
" Clear the history of connected clients\n" +
"\n" +
" --client-options\n" +
" Lists all connections and the client options\n" +
"\n" +
" -h <host>, --host <host>\n" +
" Node hostname or ip address\n" +
"\n" +
" -p <port>, --port <port>\n" +
" Remote jmx agent port number\n" +
"\n" +
" -pp, --print-port\n" +
" Operate in 4.0 mode with hosts disambiguated by port number\n" +
"\n" +
" -pw <password>, --password <password>\n" +
" Remote jmx agent password\n" +
"\n" +
" -pwf <passwordFilePath>, --password-file <passwordFilePath>\n" +
" Path to the JMX password file\n" +
"\n" +
" -u <username>, --username <username>\n" +
" Remote jmx agent username\n" +
"\n" +
"\n";
"\n" +
" -h <host>, --host <host>\n" +
" Node hostname or ip address\n" +
"\n" +
" -p <port>, --port <port>\n" +
" Remote jmx agent port number\n" +
"\n" +
" -pp, --print-port\n" +
" Operate in 4.0 mode with hosts disambiguated by port number\n" +
"\n" +
" -pw <password>, --password <password>\n" +
" Remote jmx agent password\n" +
"\n" +
" -pwf <passwordFilePath>, --password-file <passwordFilePath>\n" +
" Path to the JMX password file\n" +
"\n" +
" -u <username>, --username <username>\n" +
" Remote jmx agent username\n" +
"\n" +
"\n";
assertThat(tool.getStdout()).isEqualTo(help);
}
@Test
public void testClientStats()
{
@ -117,7 +146,7 @@ public class ClientStatsTest extends CQLTester
assertThat(stdout).contains("User Connections");
assertThat(stdout).contains("anonymous 2");
}
@Test
public void testClientStatsByProtocol()
{
@ -128,7 +157,7 @@ public class ClientStatsTest extends CQLTester
assertThat(stdout).contains("Protocol-Version IP-Address Last-Seen");
assertThat(stdout).containsPattern("[0-9]/v[0-9] +/127.0.0.1 [a-zA-Z]{3} [0-9]+, [0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}");
}
@Test
public void testClientStatsAll()
{
@ -157,7 +186,7 @@ public class ClientStatsTest extends CQLTester
assertThat(stdout).contains("User Connections");
assertThat(stdout).contains("anonymous 2");
}
@Test
public void testClientStatsClearHistory()
{
@ -166,13 +195,13 @@ public class ClientStatsTest extends CQLTester
ssLogger.addAppender(listAppender);
listAppender.start();
ToolRunner.ToolResult tool = ToolRunner.invokeNodetool("clientstats", "--clear-history");
tool.assertOnCleanExit();
String stdout = tool.getStdout();
assertThat(stdout).contains("Clearing connection history");
assertThat(listAppender.list)
.extracting(ILoggingEvent::getMessage, ILoggingEvent::getLevel)
.contains(Tuple.tuple("Cleared connection history", Level.INFO));
.extracting(ILoggingEvent::getMessage, ILoggingEvent::getLevel)
.contains(Tuple.tuple("Cleared connection history", Level.INFO));
}
}

7
test/unit/org/apache/cassandra/transport/CQLUserAuditTest.java
View File

@ -38,13 +38,13 @@ import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.AuthenticationException;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.audit.AuditEvent;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.audit.AuditLogManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.OverrideConfigurationLoader;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.diag.DiagnosticEventService;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.EmbeddedCassandraService;
@ -69,11 +69,10 @@ public class CQLUserAuditTest
config.audit_logging_options.enabled = true;
config.audit_logging_options.logger = new ParameterizedClass("DiagnosticEventAuditLogger", null);
});
CQLTester.prepareServer();
System.setProperty("cassandra.superuser_setup_delay_ms", "0");
embedded = new EmbeddedCassandraService();
embedded.start();
embedded = ServerTestUtils.startEmbeddedCassandraService();
executeAs(Arrays.asList("CREATE ROLE testuser WITH LOGIN = true AND SUPERUSER = false AND PASSWORD = 'foo'",
"CREATE ROLE testuser_nologin WITH LOGIN = false AND SUPERUSER = false AND PASSWORD = 'foo'",