extends Thread +public class MemtableCleanerThread
extends InfiniteLoopExecutor { - /** The pool we're cleaning */ - final P pool; - - /** should ensure that at least some memory has been marked reclaiming after completion */ - final Runnable cleaner; - - /** signalled whenever needsCleaning() may return true */ - final WaitQueue wait = new WaitQueue(); - - MemtableCleanerThread(P pool, Runnable cleaner) + private static class Clean
implements InterruptibleRunnable { - super(pool.getClass().getSimpleName() + "Cleaner"); - this.pool = pool; - this.cleaner = cleaner; - setDaemon(true); - } + /** The pool we're cleaning */ + final P pool; - boolean needsCleaning() - { - return pool.onHeap.needsCleaning() || pool.offHeap.needsCleaning(); - } + /** should ensure that at least some memory has been marked reclaiming after completion */ + final Runnable cleaner; - // should ONLY be called when we really think it already needs cleaning - void trigger() - { - wait.signal(); - } + /** signalled whenever needsCleaning() may return true */ + final WaitQueue wait = new WaitQueue(); - @Override - public void run() - { - while (true) + private Clean(P pool, Runnable cleaner) { - while (!needsCleaning()) + this.pool = pool; + this.cleaner = cleaner; + } + + boolean needsCleaning() + { + return pool.onHeap.needsCleaning() || pool.offHeap.needsCleaning(); + } + + @Override + public void run() throws InterruptedException + { + if (needsCleaning()) + { + cleaner.run(); + } + else { final WaitQueue.Signal signal = wait.register(); if (!needsCleaning()) - signal.awaitUninterruptibly(); + signal.await(); else signal.cancel(); } - - cleaner.run(); } } + + private final Runnable trigger; + private MemtableCleanerThread(final Clean
clean) + { + super(clean.pool.getClass().getSimpleName() + "Cleaner", clean); + this.trigger = new Runnable() + { + public void run() + { + clean.wait.signal(); + } + }; + } + + MemtableCleanerThread(P pool, Runnable cleaner) + { + this(new Clean<>(pool, cleaner)); + } + + // should ONLY be called when we really think it already needs cleaning + public void trigger() + { + trigger.run(); + } } diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java index bb858843b9..a096ec3a55 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java @@ -18,6 +18,7 @@ */ package org.apache.cassandra.utils.memory; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import org.apache.cassandra.utils.concurrent.WaitQueue; @@ -57,6 +58,12 @@ public abstract class MemtablePool } public abstract boolean needToCopyOnHeap(); + public void shutdown() throws InterruptedException + { + cleaner.shutdown(); + cleaner.awaitTermination(60, TimeUnit.SECONDS); + } + public abstract MemtableAllocator newAllocator(); /** diff --git a/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java index fae6f2a2ea..438e4116c3 100644 --- a/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java +++ b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java @@ -40,12 +40,12 @@ public class LegacyJMXProgressSupport implements ProgressListener protected static final Pattern SESSION_SUCCESS_MATCHER = Pattern.compile("Repair session .* for range .* finished"); private final AtomicLong notificationSerialNumber = new AtomicLong(); - private final ObjectName jmxObjectName; + private final String jmxObjectName; private final NotificationBroadcasterSupport broadcaster; public LegacyJMXProgressSupport(NotificationBroadcasterSupport broadcaster, - ObjectName jmxObjectName) + String jmxObjectName) { this.broadcaster = broadcaster; this.jmxObjectName = jmxObjectName; diff --git a/test/conf/logback-dtest.xml b/test/conf/logback-dtest.xml new file mode 100644 index 0000000000..b8019f6d0b --- /dev/null +++ b/test/conf/logback-dtest.xml @@ -0,0 +1,79 @@ + + + + + + + + + + + + + + ./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log + + ./build/test/logs/${cassandra.testtag}/TEST-${suitename}.log.%i.gz + 1 + 20 + + + + 20MB + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %msg%n + false + + + + + + %-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n + + + DEBUG + + + + + + + + + + + + + + + 0 + 0 + 1024 + + true + + + + + + diff --git a/test/distributed/org/apache/cassandra/distributed/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/Coordinator.java new file mode 100644 index 0000000000..5cb51a4f7c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Coordinator.java @@ -0,0 +1,79 @@ +/* + * 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; + + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +import org.apache.cassandra.cql3.CQLStatement; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.transport.Server; +import org.apache.cassandra.transport.messages.ResultMessage; +import org.apache.cassandra.utils.ByteBufferUtil; + + +public class Coordinator +{ + final Instance instance; + + public Coordinator(Instance instance) + { + this.instance = instance; + } + + private static Object[][] coordinatorExecute(String query, int consistencyLevel, Object[] bindings) + { + CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement; + List boundValues = new ArrayList<>(); + for (Object binding : bindings) + { + boundValues.add(ByteBufferUtil.objectToBytes(binding)); + } + + prepared.validate(QueryState.forInternalCalls().getClientState()); + ResultMessage res = prepared.execute(QueryState.forInternalCalls(), + QueryOptions.create(ConsistencyLevel.fromCode(consistencyLevel), + boundValues, + false, + 10, + null, + null, + Server.VERSION_4)); + + if (res != null && res.kind == ResultMessage.Kind.ROWS) + { + return RowUtil.toObjects((ResultMessage.Rows) res); + } + else + { + return new Object[][]{}; + } + } + + public Object[][] execute(String query, ConsistencyLevel consistencyLevel, Object... boundValues) + { + return instance.appliesOnInstance(Coordinator::coordinatorExecute).apply(query, consistencyLevel.code, boundValues); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java new file mode 100644 index 0000000000..ad0cacc58e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java @@ -0,0 +1,92 @@ +/* + * 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; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.cassandra.db.ConsistencyLevel; + +public class DistributedReadWritePathTest extends DistributedTestBase +{ + @Test + public void coordinatorRead() throws Throwable + { + try (TestCluster cluster = createCluster(3)) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + cluster.get(1).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)"); + cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 2, 2)"); + cluster.get(3).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 3, 3)"); + + assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", + ConsistencyLevel.ALL, + 1), + row(1, 1, 1), + row(1, 2, 2), + row(1, 3, 3)); + } + } + + @Test + public void coordinatorWrite() throws Throwable + { + try (TestCluster cluster = createCluster(3)) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + cluster.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)", + ConsistencyLevel.QUORUM); + + for (int i = 0; i < 3; i++) + { + assertRows(cluster.get(1).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"), + row(1, 1, 1)); + } + + assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", + ConsistencyLevel.QUORUM), + row(1, 1, 1)); + } + } + + @Test + public void readRepairTest() throws Throwable + { + try (TestCluster cluster = createCluster(3)) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + + cluster.get(1).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)"); + cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)"); + + assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1")); + + assertRows(cluster.coordinator().execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", + ConsistencyLevel.QUORUM), + row(1, 1, 1)); + + // Verify that data got repaired to the third node + assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"), + row(1, 1, 1)); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java b/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java new file mode 100644 index 0000000000..64ef8592ae --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/DistributedTestBase.java @@ -0,0 +1,86 @@ +/* + * 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; + +import java.util.Arrays; + +import org.junit.Assert; +import org.junit.BeforeClass; + +public class DistributedTestBase +{ + static String KEYSPACE = "distributed_test_keyspace"; + + @BeforeClass + public static void setup() + { + System.setProperty("org.apache.cassandra.disable_mbean_registration", "true"); + } + + TestCluster createCluster(int nodeCount) throws Throwable + { + TestCluster cluster = TestCluster.create(nodeCount); + cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + nodeCount + "};"); + + return cluster; + } + + public static void assertRows(Object[][] actual, Object[]... expected) + { + Assert.assertEquals(rowsNotEqualErrorMessage(actual, expected), + expected.length, actual.length); + + for (int i = 0; i < expected.length; i++) + { + Object[] expectedRow = expected[i]; + Object[] actualRow = actual[i]; + Assert.assertTrue(rowsNotEqualErrorMessage(actual, expected), + Arrays.equals(expectedRow, actualRow)); + } + } + + public static String rowsNotEqualErrorMessage(Object[][] actual, Object[][] expected) + { + return String.format("Expected: %s\nActual:%s\n", + rowsToString(expected), + rowsToString(actual)); + } + + public static String rowsToString(Object[][] rows) + { + StringBuilder builder = new StringBuilder(); + builder.append("["); + boolean isFirst = true; + for (Object[] row : rows) + { + if (isFirst) + isFirst = false; + else + builder.append(","); + builder.append(Arrays.toString(row)); + } + builder.append("]"); + return builder.toString(); + } + + public static Object[] row(Object... expected) + { + return expected; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/Instance.java new file mode 100644 index 0000000000..7564476051 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Instance.java @@ -0,0 +1,371 @@ +/* + * 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; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.File; +import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.function.BiConsumer; +import java.util.function.Function; + +import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.concurrent.SharedExecutorPool; +import org.apache.cassandra.concurrent.StageManager; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.config.Schema; +import org.apache.cassandra.cql3.CQLStatement; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.statements.ParsedStatement; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Memtable; +import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.db.commitlog.CommitLog; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.SimpleSeedProvider; +import org.apache.cassandra.locator.SimpleSnitch; +import org.apache.cassandra.net.IMessageSink; +import org.apache.cassandra.net.MessageDeliveryTask; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.net.MessageOut; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.PendingRangeCalculatorService; +import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.transport.messages.ResultMessage; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.concurrent.Ref; + +public class Instance extends InvokableInstance +{ + public final InstanceConfig config; + + public Instance(InstanceConfig config, ClassLoader classLoader) + { + super(classLoader); + this.config = config; + } + + public InetAddressAndPort getBroadcastAddress() { return callOnInstance(LegacyAdapter::getBroadcastAddressAndPort); } + + public Object[][] executeInternal(String query, Object... args) + { + return callOnInstance(() -> + { + ParsedStatement.Prepared prepared = QueryProcessor.prepareInternal(query); + ResultMessage result = prepared.statement.executeInternal(QueryProcessor.internalQueryState(), + QueryProcessor.makeInternalOptions(prepared, args)); + + if (result instanceof ResultMessage.Rows) + return RowUtil.toObjects((ResultMessage.Rows)result); + else + return null; + }); + } + + public UUID getSchemaVersion() + { + // we do not use method reference syntax here, because we need to invoke on the node-local schema instance + //noinspection Convert2MethodRef + return callOnInstance(() -> Schema.instance.getVersion()); + } + + public void schemaChange(String query) + { + runOnInstance(() -> + { + try + { + ClientState state = ClientState.forInternalCalls(); + state.setKeyspace(SystemKeyspace.NAME); + QueryState queryState = new QueryState(state); + + CQLStatement statement = QueryProcessor.parseStatement(query, queryState).statement; + statement.validate(state); + + QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList()); + statement.executeInternal(queryState, options); + } + catch (Exception e) + { + throw new RuntimeException("Error setting schema for test (query was: " + query + ")", e); + } + }); + } + + private void registerMockMessaging(TestCluster cluster) + { + BiConsumer deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); + BiConsumer deliverToInstanceIfNotFiltered = cluster.filters().filter(deliverToInstance); + + Map addressAndPortMap = new HashMap<>(); + cluster.stream().map(Instance::getBroadcastAddress).forEach(addressAndPort -> { + if (null != addressAndPortMap.put(addressAndPort.address, addressAndPort)) + throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress"); + }); + + acceptsOnInstance((BiConsumer deliver) -> + MessagingService.instance().addMessageSink(new MessageDeliverySink(deliver, addressAndPortMap::get)) + ).accept(deliverToInstanceIfNotFiltered); + } + + private static class MessageDeliverySink implements IMessageSink + { + private final BiConsumer deliver; + private final Function lookupAddressAndPort; + MessageDeliverySink(BiConsumer deliver, Function lookupAddressAndPort) + { + this.deliver = deliver; + this.lookupAddressAndPort = lookupAddressAndPort; + } + + public boolean allowOutgoingMessage(MessageOut messageOut, int id, InetAddress to) + { + try (DataOutputBuffer out = new DataOutputBuffer(1024)) + { + InetAddressAndPort from = LegacyAdapter.getBroadcastAddressAndPort(); + InetAddressAndPort toFull = lookupAddressAndPort.apply(to); + messageOut.serialize(out, MessagingService.current_version); + deliver.accept(toFull, new Message(messageOut.verb.ordinal(), out.toByteArray(), id, MessagingService.current_version, from)); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + return false; + } + + public boolean allowIncomingMessage(MessageIn message, int id) + { + // we can filter to our heart's content on the outgoing message; no need to worry about incoming + return true; + } + } + + private void receiveMessage(Message message) + { + acceptsOnInstance((Message m) -> + { + try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(m.bytes))) + { + MessageIn messageIn = MessageIn.read(in, m.version, m.id); + Runnable deliver = new MessageDeliveryTask(messageIn, m.id, System.currentTimeMillis(), false); + deliver.run(); + } + catch (Throwable t) + { + throw new RuntimeException("Exception occurred on the node " + LegacyAdapter.getBroadcastAddressAndPort(), t); + } + + }).accept(message); + } + + void launch(TestCluster cluster) + { + try + { + mkdirs(); + int id = config.num; + runOnInstance(() -> InstanceIDDefiner.instanceId = id); // for logging + + startup(); + initializeRing(cluster); + registerMockMessaging(cluster); + } + catch (Throwable t) + { + if (t instanceof RuntimeException) + throw (RuntimeException) t; + throw new RuntimeException(t); + } + } + + private void mkdirs() + { + new File(config.saved_caches_directory).mkdirs(); + new File(config.hints_directory).mkdirs(); + new File(config.commitlog_directory).mkdirs(); + for (String dir : config.data_file_directories) + new File(dir).mkdirs(); + } + + private void startup() + { + acceptsOnInstance((InstanceConfig config) -> + { + Config.setOverrideLoadConfig(() -> loadConfig(config)); + DatabaseDescriptor.setDaemonInitialized(); + DatabaseDescriptor.createAllDirectories(); + Keyspace.setInitialized(); + SystemKeyspace.persistLocalMetadata(); + }).accept(config); + } + + + public static Config loadConfig(InstanceConfig overrides) + { + Config config = new Config(); + // Defaults + config.commitlog_sync = Config.CommitLogSync.batch; + config.endpoint_snitch = SimpleSnitch.class.getName(); + config.seed_provider = new ParameterizedClass(SimpleSeedProvider.class.getName(), + Collections.singletonMap("seeds", "127.0.0.1:7010")); + // Overrides + config.partitioner = overrides.partitioner; + config.broadcast_address = overrides.broadcast_address; + config.listen_address = overrides.listen_address; + config.broadcast_rpc_address = overrides.broadcast_rpc_address; + config.rpc_address = overrides.rpc_address; + config.saved_caches_directory = overrides.saved_caches_directory; + config.data_file_directories = overrides.data_file_directories; + config.commitlog_directory = overrides.commitlog_directory; + config.concurrent_writes = overrides.concurrent_writes; + config.concurrent_counter_writes = overrides.concurrent_counter_writes; + config.concurrent_reads = overrides.concurrent_reads; + config.memtable_flush_writers = overrides.memtable_flush_writers; + config.concurrent_compactors = overrides.concurrent_compactors; + config.memtable_heap_space_in_mb = overrides.memtable_heap_space_in_mb; + config.initial_token = overrides.initial_token; + + // legacy config options we need to specify + config.commitlog_sync_batch_window_in_ms = 1.0; + return config; + } + + private void initializeRing(TestCluster cluster) + { + // This should be done outside instance in order to avoid serializing config + String partitionerName = config.partitioner; + List initialTokens = new ArrayList<>(); + List hosts = new ArrayList<>(); + List hostIds = new ArrayList<>(); + for (int i = 1 ; i <= cluster.size() ; ++i) + { + InstanceConfig config = cluster.get(i).config; + initialTokens.add(config.initial_token); + try + { + hosts.add(InetAddressAndPort.getByName(config.broadcast_address)); + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } + hostIds.add(config.hostId); + } + + runOnInstance(() -> + { + try + { + IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName); + StorageService storageService = StorageService.instance; + List tokens = new ArrayList<>(); + for (String token : initialTokens) + tokens.add(partitioner.getTokenFactory().fromString(token)); + + for (int i = 0; i < tokens.size(); i++) + { + InetAddressAndPort ep = hosts.get(i); + Gossiper.instance.initializeNodeUnsafe(ep.address, hostIds.get(i), 1); + Gossiper.instance.injectApplicationState(ep.address, + ApplicationState.TOKENS, + new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(tokens.get(i)))); + storageService.onChange(ep.address, + ApplicationState.STATUS, + new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i)))); + Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address)); + MessagingService.instance().setVersion(ep.address, MessagingService.current_version); + } + + // check that all nodes are in token metadata + for (int i = 0; i < tokens.size(); ++i) + assert storageService.getTokenMetadata().isMember(hosts.get(i).address); + } + catch (Throwable e) // UnknownHostException + { + throw new RuntimeException(e); + } + }); + } + + void shutdown() + { + runOnInstance(() -> { + Throwable error = null; + error = runAndMergeThrowable(error, + Gossiper.instance::stop, + CompactionManager.instance::forceShutdown, + CommitLog.instance::shutdownBlocking, + ColumnFamilyStore::shutdownFlushExecutor, + ColumnFamilyStore::shutdownPostFlushExecutor, + ColumnFamilyStore::shutdownReclaimExecutor, + PendingRangeCalculatorService.instance::shutdownExecutor, + Ref::shutdownReferenceReaper, + Memtable.MEMORY_POOL::shutdown, + StageManager::shutdownAndWait, + SharedExecutorPool::shutdownSharedPool, + MessagingService.instance()::shutdown, + ScheduledExecutors::shutdownAndWait); + Throwables.maybeFail(error); + }); + } + + private static Throwable runAndMergeThrowable(Throwable existing, ThrowingRunnable ... runnables) + { + for (ThrowingRunnable runnable : runnables) + { + try + { + runnable.run(); + } + catch (Throwable t) + { + existing = Throwables.merge(existing, t); + } + } + return existing; + } + + public static interface ThrowingRunnable + { + public void run() throws Throwable; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java new file mode 100644 index 0000000000..036f6b1ee4 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceClassLoader.java @@ -0,0 +1,101 @@ +/* + * 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; + +import com.google.common.base.Predicate; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.Pair; + +import java.net.URL; +import java.net.URLClassLoader; +import java.util.HashSet; +import java.util.Set; +import java.util.function.IntFunction; + +public class InstanceClassLoader extends URLClassLoader +{ + // Classes that have to be shared between instances, for configuration or returning values + private final static Class[] commonClasses = new Class[] + { + Pair.class, + InstanceConfig.class, + Message.class, + InetAddressAndPort.class, + InvokableInstance.SerializableBiConsumer.class, + InvokableInstance.SerializableBiFunction.class, + InvokableInstance.SerializableCallable.class, + InvokableInstance.SerializableConsumer.class, + InvokableInstance.SerializableFunction.class, + InvokableInstance.SerializableRunnable.class, + InvokableInstance.SerializableTriFunction.class, + InvokableInstance.InstanceFunction.class + }; + + private final int id; // for debug purposes + private final ClassLoader commonClassLoader; + private final Predicate isCommonClassName; + + InstanceClassLoader(int id, URL[] urls, Predicate isCommonClassName, ClassLoader commonClassLoader) + { + super(urls, null); + this.id = id; + this.commonClassLoader = commonClassLoader; + this.isCommonClassName = isCommonClassName; + } + + @Override + public Class loadClass(String name) throws ClassNotFoundException + { + // Do not share: + // * yaml, which is a rare exception because it does mess with loading org.cassandra...Config class instances + // * most of the rest of Cassandra classes (unless they were explicitly shared) g + if (name.startsWith("org.slf4j") || + name.startsWith("ch.qos.logback") || + name.startsWith("org.yaml") || + (name.startsWith("org.apache.cassandra") && !isCommonClassName.apply(name))) + return loadClassInternal(name); + + return commonClassLoader.loadClass(name); + } + + Class loadClassInternal(String name) throws ClassNotFoundException + { + synchronized (getClassLoadingLock(name)) + { + // First, check if the class has already been loaded + Class c = findLoadedClass(name); + + if (c == null) + c = findClass(name); + + return c; + } + } + + public static IntFunction createFactory(URLClassLoader contextClassLoader) + { + Set commonClassNames = new HashSet<>(); + for (Class k : commonClasses) + commonClassNames.add(k.getName()); + + URL[] urls = contextClassLoader.getURLs(); + return id -> new InstanceClassLoader(id, urls, commonClassNames::contains, contextClassLoader); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java new file mode 100644 index 0000000000..49c2e1f624 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceConfig.java @@ -0,0 +1,87 @@ +/* + * 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; + +import java.io.File; +import java.io.Serializable; +import java.util.UUID; + +public class InstanceConfig implements Serializable +{ + public final int num; + public final UUID hostId =java.util.UUID.randomUUID(); + public final String partitioner = "org.apache.cassandra.dht.Murmur3Partitioner"; + public final String broadcast_address; + public final String listen_address; + public final String broadcast_rpc_address; + public final String rpc_address; + public final String saved_caches_directory; + public final String[] data_file_directories; + public final String commitlog_directory; + public final String hints_directory; + public final String cdc_directory; + public final int concurrent_writes = 2; + public final int concurrent_counter_writes = 2; + public final int concurrent_materialized_view_writes = 2; + public final int concurrent_reads = 2; + public final int memtable_flush_writers = 1; + public final int concurrent_compactors = 1; + public final int memtable_heap_space_in_mb = 10; + public final String initial_token; + + private InstanceConfig(int num, + String broadcast_address, + String listen_address, + String broadcast_rpc_address, + String rpc_address, + String saved_caches_directory, + String[] data_file_directories, + String commitlog_directory, + String hints_directory, + String cdc_directory, + String initial_token) + { + this.num = num; + this.broadcast_address = broadcast_address; + this.listen_address = listen_address; + this.broadcast_rpc_address = broadcast_rpc_address; + this.rpc_address = rpc_address; + this.saved_caches_directory = saved_caches_directory; + this.data_file_directories = data_file_directories; + this.commitlog_directory = commitlog_directory; + this.hints_directory = hints_directory; + this.cdc_directory = cdc_directory; + this.initial_token = initial_token; + } + + public static InstanceConfig generate(int nodeNum, File root, String token) + { + return new InstanceConfig(nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + "127.0.0." + nodeNum, + String.format("%s/node%d/saved_caches", root, nodeNum), + new String[] { String.format("%s/node%d/data", root, nodeNum) }, + String.format("%s/node%d/commitlog", root, nodeNum), + String.format("%s/node%d/hints", root, nodeNum), + String.format("%s/node%d/cdc", root, nodeNum), + token); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java b/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java new file mode 100644 index 0000000000..11677487fa --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InstanceIDDefiner.java @@ -0,0 +1,38 @@ +/* + * 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; + +import ch.qos.logback.core.PropertyDefinerBase; + +/** + * Used by logback to find/define property value, see logback-dtest.xml + */ +public class InstanceIDDefiner extends PropertyDefinerBase +{ + // Instantiated per classloader, set by Instance + public static int instanceId = -1; + + public String getPropertyValue() + { + if (instanceId == -1) + return ""; + else + return "INSTANCE" + instanceId; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java new file mode 100644 index 0000000000..f646ae1f8f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/InvokableInstance.java @@ -0,0 +1,133 @@ +/* + * 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; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.concurrent.Callable; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; + +public abstract class InvokableInstance +{ + private final ClassLoader classLoader; + private final Method deserializeOnInstance; + + public InvokableInstance(ClassLoader classLoader) + { + this.classLoader = classLoader; + try + { + this.deserializeOnInstance = classLoader.loadClass(InvokableInstance.class.getName()).getDeclaredMethod("deserializeOneObject", byte[].class); + } + catch (ClassNotFoundException | NoSuchMethodException e) + { + throw new RuntimeException(e); + } + } + + public interface SerializableCallable extends Callable, Serializable { public T call(); } + public SerializableCallable callsOnInstance(SerializableCallable call) { return (SerializableCallable) transferOneObject(call); } + public T callOnInstance(SerializableCallable call) { return callsOnInstance(call).call(); } + + public interface SerializableRunnable extends Runnable, Serializable {} + public SerializableRunnable runsOnInstance(SerializableRunnable run) { return (SerializableRunnable) transferOneObject(run); } + public void runOnInstance(SerializableRunnable run) { runsOnInstance(run).run(); } + + public interface SerializableConsumer extends Consumer, Serializable {} + public SerializableConsumer acceptsOnInstance(SerializableConsumer consumer) { return (SerializableConsumer) transferOneObject(consumer); } + + public interface SerializableBiConsumer extends BiConsumer, Serializable {} + public SerializableBiConsumer acceptsOnInstance(SerializableBiConsumer consumer) { return (SerializableBiConsumer) transferOneObject(consumer); } + + public interface SerializableFunction extends Function, Serializable {} + public SerializableFunction appliesOnInstance(SerializableFunction f) { return (SerializableFunction) transferOneObject(f); } + + public interface SerializableBiFunction extends BiFunction, Serializable {} + public SerializableBiFunction appliesOnInstance(SerializableBiFunction f) { return (SerializableBiFunction) transferOneObject(f); } + + public interface SerializableTriFunction extends Serializable + { + O apply(I1 i1, I2 i2, I3 i3); + } + public SerializableTriFunction appliesOnInstance(SerializableTriFunction f) { return (SerializableTriFunction) transferOneObject(f); } + + public interface InstanceFunction extends SerializableBiFunction {} + + // E must be a functional interface, and lambda must be implemented by a lambda function + public E invokesOnInstance(E lambda) + { + return (E) transferOneObject(lambda); + } + + public Object transferOneObject(Object object) + { + byte[] bytes = serializeOneObject(object); + try + { + Object onInstance = deserializeOnInstance.invoke(null, bytes); + if (onInstance.getClass().getClassLoader() != classLoader) + throw new IllegalStateException(onInstance + " seemingly from wrong class loader: " + onInstance.getClass().getClassLoader() + ", but expected " + classLoader); + + return onInstance; + } + catch (IllegalAccessException | InvocationTargetException e) + { + throw new RuntimeException(e); + } + } + + private byte[] serializeOneObject(Object object) + { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos)) + { + oos.writeObject(object); + oos.close(); + return baos.toByteArray(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @SuppressWarnings("unused") // called through method invocation + public static Object deserializeOneObject(byte[] bytes) + { + try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + ObjectInputStream ois = new ObjectInputStream(bais);) + { + return ois.readObject(); + } + catch (IOException | ClassNotFoundException e) + { + throw new RuntimeException(e); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java b/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java new file mode 100644 index 0000000000..1ff88eda75 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/LegacyAdapter.java @@ -0,0 +1,42 @@ +/* + * 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; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.FBUtilities; + +import java.net.InetAddress; + +public class LegacyAdapter +{ + private static final InetAddressAndPort broadcastAddressAndPort; + static + { + InetAddress address = FBUtilities.getBroadcastAddress(); + int port = DatabaseDescriptor.getStoragePort(); + broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(address, port); + } + + public static InetAddressAndPort getBroadcastAddressAndPort() + { + return broadcastAddressAndPort; + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/Message.java b/test/distributed/org/apache/cassandra/distributed/Message.java new file mode 100644 index 0000000000..b5492a2a2c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/Message.java @@ -0,0 +1,41 @@ +/* + * 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; + +import org.apache.cassandra.locator.InetAddressAndPort; + +// a container for simplifying the method signature for per-instance message handling/delivery +public class Message +{ + public final int verb; + public final byte[] bytes; + public final int id; + public final int version; + public final InetAddressAndPort from; + + public Message(int verb, byte[] bytes, int id, int version, InetAddressAndPort from) + { + this.verb = verb; + this.bytes = bytes; + this.id = id; + this.version = version; + this.from = from; + } +} + diff --git a/test/distributed/org/apache/cassandra/distributed/MessageFilters.java b/test/distributed/org/apache/cassandra/distributed/MessageFilters.java new file mode 100644 index 0000000000..47eb52bcbd --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/MessageFilters.java @@ -0,0 +1,175 @@ +/* + * 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; + +import java.util.Arrays; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.function.BiConsumer; + +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessagingService; + +public class MessageFilters +{ + private final TestCluster cluster; + private final Set filters = new CopyOnWriteArraySet<>(); + + public MessageFilters(TestCluster cluster) + { + this.cluster = cluster; + } + + BiConsumer filter(BiConsumer applyIfNotFiltered) + { + return (toAddress, message) -> + { + int from = cluster.get(message.from).config.num; + int to = cluster.get(toAddress).config.num; + int verb = message.verb; + for (Filter filter : filters) + { + if (filter.matches(from, to, verb)) + return; + } + + applyIfNotFiltered.accept(toAddress, message); + }; + } + + public class Filter + { + final int[] from; + final int[] to; + final int[] verbs; + + Filter(int[] from, int[] to, int[] verbs) + { + if (from != null) + { + from = from.clone(); + Arrays.sort(from); + } + if (to != null) + { + to = to.clone(); + Arrays.sort(to); + } + if (verbs != null) + { + verbs = verbs.clone(); + Arrays.sort(verbs); + } + this.from = from; + this.to = to; + this.verbs = verbs; + } + + public int hashCode() + { + return (from == null ? 0 : Arrays.hashCode(from)) + + (to == null ? 0 : Arrays.hashCode(to)) + + (verbs == null ? 0 : Arrays.hashCode(verbs)); + } + + public boolean equals(Object that) + { + return that instanceof Filter && equals((Filter) that); + } + + public boolean equals(Filter that) + { + return Arrays.equals(from, that.from) + && Arrays.equals(to, that.to) + && Arrays.equals(verbs, that.verbs); + } + + public boolean matches(int from, int to, int verb) + { + return (this.from == null || Arrays.binarySearch(this.from, from) >= 0) + && (this.to == null || Arrays.binarySearch(this.to, to) >= 0) + && (this.verbs == null || Arrays.binarySearch(this.verbs, verb) >= 0); + } + + public Filter restore() + { + filters.remove(this); + return this; + } + + public Filter drop() + { + filters.add(this); + return this; + } + } + + public class Builder + { + int[] from; + int[] to; + int[] verbs; + + private Builder(int[] verbs) + { + this.verbs = verbs; + } + + public Builder from(int ... nums) + { + from = nums; + return this; + } + + public Builder to(int ... nums) + { + to = nums; + return this; + } + + public Filter ready() + { + return new Filter(from, to, verbs); + } + + public Filter drop() + { + return ready().drop(); + } + } + + public Builder verbs(MessagingService.Verb ... verbs) + { + int[] ids = new int[verbs.length]; + for (int i = 0 ; i < verbs.length ; ++i) + ids[i] = verbs[i].ordinal(); + return new Builder(ids); + } + + public Builder allVerbs() + { + return new Builder(null); + } + + public void reset() + { + filters.clear(); + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/RowUtil.java new file mode 100644 index 0000000000..bce896d784 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/RowUtil.java @@ -0,0 +1,47 @@ +/* + * 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; + +import java.nio.ByteBuffer; +import java.util.List; + +import org.apache.cassandra.cql3.ColumnSpecification; +import org.apache.cassandra.transport.messages.ResultMessage; + +public class RowUtil +{ + public static Object[][] toObjects(ResultMessage.Rows rows) + { + Object[][] result = new Object[rows.result.rows.size()][]; + List specs = rows.result.metadata.names; + for (int i = 0; i < rows.result.rows.size(); i++) + { + List row = rows.result.rows.get(i); + result[i] = new Object[row.size()]; + for (int j = 0; j < row.size(); j++) + { + ByteBuffer bb = row.get(j); + + if (bb != null) + result[i][j] = specs.get(j).type.getSerializer().deserialize(bb); + } + } + return result; + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/TestCluster.java b/test/distributed/org/apache/cassandra/distributed/TestCluster.java new file mode 100644 index 0000000000..20a4ab259a --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/TestCluster.java @@ -0,0 +1,277 @@ +/* + * 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; + +import java.io.File; +import java.io.IOException; +import java.net.URLClassLoader; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import java.util.function.IntFunction; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import com.google.common.collect.Sets; + +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.FBUtilities; + +/** + * TestCluster creates, initializes and manages Cassandra instances ({@link Instance}. + * + * All instances created under the same cluster will have a shared ClassLoader that'll preload + * common classes required for configuration and communication (byte buffers, primitives, config + * objects etc). Shared classes are listed in {@link InstanceClassLoader#commonClasses}. + * + * Each instance has its own class loader that will load logging, yaml libraries and all non-shared + * Cassandra package classes. The rule of thumb is that we'd like to have all Cassandra-specific things + * (unless explitily shared through the common classloader) on a per-classloader basis in order to + * allow creating more than one instance of DatabaseDescriptor and other Cassandra singletones. + * + * All actions (reading, writing, schema changes, etc) are executed by serializing lambda/runnables, + * transferring them to instance-specific classloaders, deserializing and running them there. Most of + * the things can be simply captured in closure or passed through `apply` method of the wrapped serializable + * function/callable. You can use {@link InvokableInstance#{applies|runs|consumes}OnInstance} for executing + * code on specific instance. + * + * Each instance has its own logger. Each instance log line will contain INSTANCE{instance_id}. + * + * As of today, messaging is faked by hooking into MessagingService, so we're not using usual Cassandra + * handlers for internode to have more control over it. Messaging is wired by passing verbs manually. + * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}. + */ +public class TestCluster implements AutoCloseable +{ + private static ExecutorService exec = Executors.newCachedThreadPool(new NamedThreadFactory("cluster-async-tasks")); + + private final File root; + private final List instances; + private final Coordinator coordinator; + private final Map instanceMap; + private final MessageFilters filters; + + private TestCluster(File root, List instances) + { + this.root = root; + this.instances = instances; + this.instanceMap = new HashMap<>(); + this.coordinator = new Coordinator(instances.get(0)); + this.filters = new MessageFilters(this); + } + + void launch() + { + FBUtilities.waitOnFutures(instances.stream() + .map(i -> exec.submit(() -> i.launch(this))) + .collect(Collectors.toList()) + ); + for (Instance instance : instances) + instanceMap.put(instance.getBroadcastAddress(), instance); + } + + public int size() + { + return instances.size(); + } + + public Coordinator coordinator() + { + return coordinator; + } + + /** + * WARNING: we index from 1 here, for consistency with inet address! + */ + public Instance get(int idx) + { + return instances.get(idx - 1); + } + + public Stream stream() { return instances.stream(); } + + public Instance get(InetAddressAndPort addr) + { + return instanceMap.get(addr); + } + + MessageFilters filters() + { + return filters; + } + + MessageFilters.Builder verbs(MessagingService.Verb ... verbs) + { + return filters.verbs(verbs); + } + + public void disableAutoCompaction(String keyspace) + { + for (Instance instance : instances) + { + instance.runOnInstance(() -> { + for (ColumnFamilyStore cs : Keyspace.open(keyspace).getColumnFamilyStores()) + cs.disableAutoCompaction(); + }); + } + } + + public void schemaChange(String query) + { + try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) + { + // execute the schema change + coordinator().execute(query, ConsistencyLevel.ALL); + monitor.waitForAgreement(); + } + } + + /** + * Will wait for a schema change AND agreement that occurs after it is created + * (and precedes the invocation to waitForAgreement) + * + * Works by simply checking if all UUIDs agree after any schema version change event, + * so long as the waitForAgreement method has been entered (indicating the change has + * taken place on the coordinator) + * + * This could perhaps be made a little more robust, but this should more than suffice. + */ + public class SchemaChangeMonitor implements AutoCloseable + { + public SchemaChangeMonitor() {} + + @Override + public void close() { } + + public void waitForAgreement() + { + long start = System.nanoTime(); + while (1 != instances.stream().map(Instance::getSchemaVersion).distinct().count()) + { + if (System.nanoTime() - start > TimeUnit.MINUTES.toNanos(1L)) + throw new IllegalStateException("Schema agreement not reached"); + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L)); + } + } + } + + public void schemaChange(String statement, int instance) + { + get(instance).schemaChange(statement); + } + + public static TestCluster create(int nodeCount) throws Throwable + { + return create(nodeCount, Files.createTempDirectory("dtests").toFile()); + } + + public static TestCluster create(int nodeCount, File root) + { + root.mkdirs(); + setupLogging(root); + + IntFunction classLoaderFactory = InstanceClassLoader.createFactory( + (URLClassLoader) Thread.currentThread().getContextClassLoader()); + List instances = new ArrayList<>(); + long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); + for (int i = 0 ; i < nodeCount ; ++i) + { + InstanceConfig instanceConfig = InstanceConfig.generate(i + 1, root, String.valueOf(token)); + instances.add(new Instance(instanceConfig, classLoaderFactory.apply(i + 1))); + token += increment; + } + + TestCluster cluster = new TestCluster(root, instances); + cluster.launch(); + return cluster; + } + + private static void setupLogging(File root) + { + try + { + String testConfPath = "test/conf/logback-dtest.xml"; + Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml"); + if (!logConfPath.toFile().exists()) + { + Files.copy(new File(testConfPath).toPath(), + logConfPath); + } + System.setProperty("logback.configurationFile", "file://" + logConfPath); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @Override + public void close() + { + List> futures = instances.stream() + .map(i -> exec.submit(i::shutdown)) + .collect(Collectors.toList()); + +// withThreadLeakCheck(futures); + + // Make sure to only delete directory when threads are stopped + exec.submit(() -> { + FBUtilities.waitOnFutures(futures); + FileUtils.deleteRecursive(root); + }); + } + + // We do not want this check to run every time until we fix problems with tread stops + private void withThreadLeakCheck(List> futures) + { + FBUtilities.waitOnFutures(futures); + + Set threadSet = Thread.getAllStackTraces().keySet(); + threadSet = Sets.difference(threadSet, Collections.singletonMap(Thread.currentThread(), null).keySet()); + if (!threadSet.isEmpty()) + { + for (Thread thread : threadSet) + { + System.out.println(thread); + System.out.println(Arrays.toString(thread.getStackTrace())); + } + throw new RuntimeException(String.format("Not all threads have shut down. %d threads are still running: %s", threadSet.size(), threadSet)); + } + } + +} +