cassandra-injvm-dtest

# Apache Cassandra In-JVM DTest Writer ## Overview Write comprehensive distributed tests for Apache Cassandra using the in-JVM dtest framework. This framework enables multi-node cluster simulation within a single JVM using isolated classloaders, providing fast and reliable integration testing without actual network communication. ## Core Concepts ### Framework Architecture The in-JVM dtest framework uses **classloader isolation** to run multiple Cassandra instances in one JVM: - **Shared ClassLoader**: Common classes (primitives, APIs, config objects, byte buffers) - **Per-Instance ClassLoader**: Isolated Cassandra internals (DatabaseDescriptor, singletons, all C* code) - **IsolatedExecutor**: Marshals serialized lambdas across classloader boundaries - **Message Interception**: Inter-node messaging hooked at MessagingService level (no actual networking) ### Key Classes | Class | Purpose | Location | |-------|---------|----------| | `Cluster` | Main entry point for creating clusters | `org.apache.cassandra.distributed.Cluster` | | `AbstractCluster` | Core cluster management | `org.apache.cassandra.distributed.impl.AbstractCluster` | | `Instance` | Individual Cassandra node | `org.apache.cassandra.distributed.impl.Instance` | | `InstanceConfig` | Node configuration | `org.apache.cassandra.distributed.impl.InstanceConfig` | | `ClusterUtils` | Utility methods | `org.apache.cassandra.distributed.shared.ClusterUtils` | | `ICoordinator` | Query execution interface | API package | | `MessageFilters` | Message filtering/dropping | `org.apache.cassandra.distributed.impl.MessageFilters` | ## Quick Start ### Basic Cluster Setup ```java import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; // Create and start a 3-node cluster try (Cluster cluster = Cluster.build(3).start()) { // Create keyspace cluster.schemaChange("CREATE KEYSPACE test WITH replication = " + "{'class': 'SimpleStrategy', 'replication_factor': 2}"); // Create table cluster.schemaChange("CREATE TABLE test.users (id int PRIMARY KEY, name text)"); // Execute query cluster.coordinator(1).execute("INSERT INTO test.users (id, name) VALUES (?, ?)", ConsistencyLevel.QUORUM, 1, "Alice"); // Read data Object[][] rows = cluster.coordinator(1).execute( "SELECT * FROM test.users WHERE id = ?", ConsistencyLevel.QUORUM, 1); } ``` ### Test Base Class Pattern ```java import org.apache.cassandra.distributed.test.TestBaseImpl; public class MyDTest extends TestBaseImpl { private static Cluster cluster; @BeforeClass public static void setupCluster() throws IOException { cluster = init(Cluster.build(3).start(), 2); // RF=2 } @AfterClass public static void teardownCluster() { if (cluster != null) cluster.close(); } @Test public void testSomething() { // Test implementation } } ``` ## Cluster Creation Patterns ### Pattern 1: Simple Single-Node Cluster ```java Cluster cluster = Cluster.build(1).start(); ``` ### Pattern 2: Multi-Node Cluster with Configuration ```java Cluster cluster = Cluster.build(3) .withConfig(config -> config .set("num_tokens", "256") .set("concurrent_reads", "64") .set("concurrent_writes", "64")) .start(); ``` ### Pattern 3: Multi-Datacenter Cluster ```java Cluster cluster = builder() .withRacks(2, 1, 3) // 2 DCs, 1 rack per DC, 3 nodes per rack .start(); // Or with explicit DC/rack naming Cluster cluster = builder() .withDC("dc1", 3) .withDC("dc2", 3) .start(); ``` ### Pattern 4: Cluster with Features ```java import static org.apache.cassandra.distributed.api.Feature.*; Cluster cluster = Cluster.build() .withNodes(3) .withConfig(config -> config .with(GOSSIP) // Enable gossip protocol .with(NETWORK) // Enable inter-node networking .with(NATIVE_PROTOCOL) // Enable CQL protocol .with(JMX)) // Enable JMX .start(); ``` ### Pattern 5: Custom Token Distribution ```java import org.apache.cassandra.distributed.api.TokenSupplier; Cluster cluster = Cluster.build() .withNodes(3) .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(3)) .start(); ``` ### Pattern 6: Dynamic Port Allocation ```java Cluster cluster = Cluster.build(3) .withDynamicPortAllocation(true) .start(); ``` ## Configuration ### Available Configuration Parameters Configuration is applied via `config.set(key, value)` or `config.with(Feature)`: #### Network Configuration ```java config.set("broadcast_address", "127.0.0.1") .set("listen_address", "127.0.0.1") .set("rpc_address", "127.0.0.1") .set("storage_port", "7000") .set("native_transport_port", "9042") .set("jmx_port", "7199"); ``` #### Token Configuration ```java config.set("num_tokens", "256") .set("initial_token", "0"); ``` #### Memory Configuration ```java config.set("memtable_heap_space", "512MiB") .set("key_cache_size", "50MiB") .set("counter_cache_size", "50MiB") .set("row_cache_size", "0MiB"); ``` #### Compaction & Concurrency ```java config.set("concurrent_compactors", "4") .set("concurrent_reads", "32") .set("concurrent_writes", "32") .set("concurrent_counter_writes", "32"); ``` #### Commitlog Configuration ```java config.set("commitlog_sync", "periodic") .set("commitlog_sync_period_in_ms", "10000") .set("commitlog_segment_size", "32MiB"); ``` #### Authentication & Authorization ```java config.set("authenticator", "PasswordAuthenticator") .set("authorizer", "CassandraAuthorizer") .set("role_manager", "CassandraRoleManager"); ``` #### Accord Configuration (CEP-15) ```java config.set("accord.enabled", "true") .set("accord.journal_directory", "/path/to/journal"); ``` #### Data Directories ```java // Data directories are arrays config.set("data_file_directories", new String[]{"/path/to/data"}) .set("commitlog_directory", "/path/to/commitlog") .set("hints_directory", "/path/to/hints") .set("saved_caches_directory", "/path/to/saved_caches"); ``` ## Instance Lifecycle Operations ### Starting Instances ```java // Start with default settings instance.startup(); // Start with system properties ClusterUtils.start(instance, properties -> { properties.set(RING_DELAY, "5000"); properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, "5000"); }); ``` ### Stopping Instances ```java // Graceful shutdown (blocking) ClusterUtils.stopUnchecked(instance); // Abrupt shutdown (simulates kill -9) ClusterUtils.stopAbrupt(cluster, instance); // Stop all instances ClusterUtils.stopAll(cluster); // Async shutdown Future<?> shutdownFuture = instance.shutdown(); shutdownFuture.get(); // wait for completion ``` ### Restarting Instances ```java // Simple restart ClusterUtils.restartUnchecked(instance); // Restart with different configuration instance.shutdown().get(); instance.config().set("concurrent_reads", "64"); instance.startup(); ``` ### Adding Instances ```java // Add instance with same config as existing nodes IInstance newInstance = ClusterUtils.addInstance(cluster); // Add instance with custom config IInstance newInstance = ClusterUtils.addInstance(cluster, config -> { config.set("auto_bootstrap", true); }); // Add instance to specific DC/rack IInstance newInstance = ClusterUtils.addInstance(cluster, "dc1", "rack1", config -> { config.set("num_tokens", "256"); }); // Start the new instance newInstance.startup(); ``` ### Replacing Instances ```java // Replace a failed node IInstance replacement = ClusterUtils.replaceHostAndStart(cluster, failedInstance); // Replace with custom properties IInstance replacement = ClusterUtils.replaceHostAndStart( cluster, failedInstance, (inst, properties) -> { properties.set(RING_DELAY, "5000"); } ); ``` ## Running Code on Instances ### Pattern 1: Run Without Return Value ```java cluster.get(1).runOnInstance(() -> { StorageService.instance.forceKeyspaceFlush("system"); }); ``` ### Pattern 2: Get Return Value (Sync) ```java String version = cluster.get(1).callOnInstance(() -> { return FBUtilities.getReleaseVersionString(); }); ``` ### Pattern 3: Get Return Value (Async) ```java Future<String> future = cluster.get(1).asyncCallOnInstance(() -> { return FBUtilities.getReleaseVersionString(); }); String version = future.get(); ``` ### Pattern 4: Pass Parameters ```java // Parameters must be serializable int result = cluster.get(1).callOnInstance(() -> { // Access Cassandra internals here return ClusterMetadata.current().directory.allAddresses().size(); }); ``` ### Pattern 5: Instance-to-Instance Operations ```java // Get node ID from one instance, use it on another int nodeId = ClusterUtils.getNodeId(cluster.get(1), cluster.get(2)); NodeId id = new NodeId(nodeId); cluster.get(2).runOnInstance(() -> { StorageService.instance.cancelInProgressSequences(id); }); ``` ## Query Execution ### Basic Queries ```java // Execute without results cluster.coordinator(1).execute( "INSERT INTO test.users (id, name) VALUES (?, ?)", ConsistencyLevel.QUORUM, 1, "Alice" ); // Execute with results Object[][] rows = cluster.coordinator(1).execute( "SELECT * FROM test.users WHERE id = ?", ConsistencyLevel.QUORUM, 1 ); ``` ### Using Query Results ```java import org.apache.cassandra.distributed.api.SimpleQueryResult; SimpleQueryResult result = cluster.coordinator(1).executeWithResult( "SELECT * FROM test.users", ConsistencyLevel.QUORUM ); while (result.hasNext()) { Row row = result.next(); int id = row.getInteger("id"); String name = row.getString("name"); System.out.println("User: " + id + ", " + name); } ``` ### Pagination ```java SimpleQueryResult result = cluster.coordinator(1).executeWithPagingWithResult( "SELECT * FROM test.users", ConsistencyLevel.QUORUM, 100 // page size ); ``` ### Async Execution ```java Future<SimpleQueryResult> future = cluster.coordinator(1).asyncExecuteWithResult( "SELECT * FROM test.users",