From db5832120f0fe9ad799ca0b5b16bbcf6f7404580 Mon Sep 17 00:00:00 2001 From: Jon Meredith Date: Thu, 22 Oct 2020 10:54:55 -0700 Subject: [PATCH] In-JVM dtest - modify schema with stopped nodes and use yaml fragments for config patch by Jon Meredith; reviewed by Alex Petrov, David Capwell, Dinesh Joshi, Yifan Cai for CASSANDRA-16152 --- .../cassandra/config/ParameterizedClass.java | 5 + .../config/YamlConfigurationLoader.java | 63 +++++++++++++ .../distributed/impl/AbstractCluster.java | 44 +++++++-- .../cassandra/distributed/impl/Instance.java | 19 ++-- .../distributed/impl/InstanceConfig.java | 49 ++-------- .../distributed/test/JVMDTestTest.java | 91 ++++++++++++++++++- .../config/YamlConfigurationLoaderTest.java | 54 +++++++++++ 7 files changed, 267 insertions(+), 58 deletions(-) create mode 100644 test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java diff --git a/src/java/org/apache/cassandra/config/ParameterizedClass.java b/src/java/org/apache/cassandra/config/ParameterizedClass.java index 6b7af63cab..7ec2c18ef0 100644 --- a/src/java/org/apache/cassandra/config/ParameterizedClass.java +++ b/src/java/org/apache/cassandra/config/ParameterizedClass.java @@ -28,6 +28,11 @@ public class ParameterizedClass public String class_name; public Map parameters; + public ParameterizedClass() + { + // for snakeyaml + } + public ParameterizedClass(String class_name, Map parameters) { this.class_name = class_name; diff --git a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java index dc691c4dc5..07b149cbcb 100644 --- a/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java +++ b/src/java/org/apache/cassandra/config/YamlConfigurationLoader.java @@ -24,11 +24,15 @@ import java.io.InputStream; import java.io.IOException; import java.net.URL; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.TreeMap; import com.google.common.base.Joiner; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; import com.google.common.io.ByteStreams; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,10 +40,14 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.exceptions.ConfigurationException; import org.yaml.snakeyaml.TypeDescription; import org.yaml.snakeyaml.Yaml; +import org.yaml.snakeyaml.composer.Composer; +import org.yaml.snakeyaml.constructor.Constructor; +import org.yaml.snakeyaml.constructor.CustomClassLoaderConstructor; import org.yaml.snakeyaml.error.YAMLException; import org.yaml.snakeyaml.introspector.MissingProperty; import org.yaml.snakeyaml.introspector.Property; import org.yaml.snakeyaml.introspector.PropertyUtils; +import org.yaml.snakeyaml.nodes.Node; public class YamlConfigurationLoader implements ConfigurationLoader { @@ -119,6 +127,61 @@ public class YamlConfigurationLoader implements ConfigurationLoader } } + @SuppressWarnings("unchecked") //getSingleData returns Object, not T + public static T fromMap(Map map, Class klass) + { + Constructor constructor = new YamlConfigurationLoader.CustomConstructor(klass, klass.getClassLoader()); + YamlConfigurationLoader.MissingPropertiesChecker propertiesChecker = new YamlConfigurationLoader.MissingPropertiesChecker(); + constructor.setPropertyUtils(propertiesChecker); + Yaml yaml = new Yaml(constructor); + Node node = yaml.represent(map); + constructor.setComposer(new Composer(null, null) + { + @Override + public Node getSingleNode() + { + return node; + } + }); + return (T) constructor.getSingleData(klass); + } + + static class CustomConstructor extends CustomClassLoaderConstructor + { + CustomConstructor(Class theRoot, ClassLoader classLoader) + { + super(theRoot, classLoader); + + TypeDescription seedDesc = new TypeDescription(ParameterizedClass.class); + seedDesc.putMapPropertyType("parameters", String.class, String.class); + addTypeDescription(seedDesc); + } + + @Override + protected List createDefaultList(int initSize) + { + return Lists.newCopyOnWriteArrayList(); + } + + @Override + protected Map createDefaultMap() + { + return Maps.newConcurrentMap(); + } + + @Override + protected Set createDefaultSet(int initSize) + { + return Sets.newConcurrentHashSet(); + } + + @Override + protected Set createDefaultSet() + { + return Sets.newConcurrentHashSet(); + } + } + private static class MissingPropertiesChecker extends PropertyUtils { private final Set missingProperties = new HashSet<>(); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index cb55d3e0e3..5f2b6248c8 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -330,7 +330,7 @@ public abstract class AbstractCluster implements ICluster implements ICluster !i.isShutdown()).findFirst().orElseThrow( + () -> new IllegalStateException("All instances are shutdown")); + } + public int size() { return instances.size(); @@ -459,11 +465,31 @@ public abstract class AbstractCluster implements ICluster { + schemaChange(query, false); + } + + /** + * Change the schema of the cluster, tolerating stopped nodes. N.B. the schema + * will not automatically be updated when stopped nodes are restarted, individual tests need to + * re-synchronize somehow (by gossip or some other mechanism). + * @param query Schema altering statement + */ + public void schemaChangeIgnoringStoppedInstances(String query) + { + schemaChange(query, true); + } + + private void schemaChange(String query, boolean ignoreStoppedInstances) + { + I instance = ignoreStoppedInstances ? getFirstRunningInstance() : get(1); + + instance.sync(() -> { try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) { // execute the schema change - coordinator(1).execute(query, ConsistencyLevel.ALL); + instance.coordinator().execute(query, ConsistencyLevel.ALL); + if (ignoreStoppedInstances) + monitor.ignoreStoppedInstances(); monitor.waitForCompletion(); } }).run(); @@ -498,16 +524,23 @@ public abstract class AbstractCluster implements ICluster instanceFilter; volatile boolean changed; public ChangeMonitor(long timeOut, TimeUnit timeoutUnit) { this.timeOut = timeOut; this.timeoutUnit = timeoutUnit; + this.instanceFilter = i -> true; this.cleanup = new ArrayList<>(instances.size()); this.completed = new SimpleCondition(); } + public void ignoreStoppedInstances() + { + instanceFilter = instanceFilter.and(i -> !i.isShutdown()); + } + protected void signal() { if (changed && isCompleted()) @@ -539,8 +572,7 @@ public abstract class AbstractCluster implements ICluster cleanup.add(startPolling(instance))); } protected abstract IListen.Cancel startPolling(IInstance instance); @@ -575,7 +607,7 @@ public abstract class AbstractCluster implements ICluster, Function> mapper = new HashMap, Function>() {{ - this.put(IInstanceConfig.ParameterizedClass.class, (obj) -> { - IInstanceConfig.ParameterizedClass pc = (IInstanceConfig.ParameterizedClass) obj; - return new org.apache.cassandra.config.ParameterizedClass(pc.class_name, pc.parameters); - }); - }}; - public final IInstanceConfig config; // should never be invoked directly, so that it is instantiated on other class loader; @@ -270,6 +264,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress toAddress) { + if (isShutdown()) + return false; + // Port is not passed in, so take a best guess at the destination port from this instance IInstance to = cluster.get(NetworkTopology.addressAndPort(toAddress, instance.config().broadcastAddress().getPort())); @@ -282,6 +279,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance public boolean allowIncomingMessage(MessageIn message, int id) { + if (isShutdown()) + return false; + // Port is not passed in, so take a best guess at the destination port from this instance IInstance from = cluster.get(NetworkTopology.addressAndPort(message.from, instance.config().broadcastAddress().getPort())); @@ -624,9 +624,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance private static Config loadConfig(IInstanceConfig overrides) { - Config config = new Config(); - overrides.propagate(config, mapper); - return config; + Map params = ((InstanceConfig) overrides).getParams(); + return YamlConfigurationLoader.fromMap(params, Config.class); } private void initializeRing(ICluster cluster) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 4e8a782a19..861e2eac20 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -30,16 +30,19 @@ import java.util.TreeMap; import java.util.UUID; import java.util.function.Function; +import org.apache.cassandra.config.YamlConfigurationLoader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.shared.Shared; import org.apache.cassandra.distributed.shared.Versions; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.SimpleSeedProvider; +@Shared public class InstanceConfig implements IInstanceConfig { private static final Object NULL = new Object(); @@ -198,8 +201,7 @@ public class InstanceConfig implements IInstanceConfig public void propagate(Object writeToConfig, Map, Function> mapping) { - for (Map.Entry e : params.entrySet()) - propagate(writeToConfig, e.getKey(), e.getValue(), mapping); + throw new IllegalStateException("In-JVM dtests no longer support propagate"); } public void validate() @@ -208,44 +210,6 @@ public class InstanceConfig implements IInstanceConfig throw new IllegalArgumentException("In-JVM dtests do not support vnodes as of now."); } - private void propagate(Object writeToConfig, String fieldName, Object value, Map, Function> mapping) - { - if (value == NULL) - value = null; - - if (mapping != null && mapping.containsKey(value.getClass())) - value = mapping.get(value.getClass()).apply(value); - - Class configClass = writeToConfig.getClass(); - Field valueField; - try - { - valueField = configClass.getDeclaredField(fieldName); - } - catch (NoSuchFieldException e) - { - logger.warn("No such field: {} in config class {}", fieldName, configClass); - return; - } - - if (valueField.getType().isEnum() && value instanceof String) - { - String test = (String) value; - value = Arrays.stream(valueField.getType().getEnumConstants()) - .filter(e -> ((Enum)e).name().equals(test)) - .findFirst() - .get(); - } - try - { - valueField.set(writeToConfig, value); - } - catch (IllegalAccessException | IllegalArgumentException e) - { - throw new IllegalStateException(e); - } - } - public Object get(String name) { return params.get(name); @@ -261,6 +225,11 @@ public class InstanceConfig implements IInstanceConfig return (String)params.get(name); } + public Map getParams() + { + return params; + } + public static InstanceConfig generate(int nodeNum, String ipAddress, NetworkTopology networkTopology, File root, String token, String seedIp, int datadirCount) { return new InstanceConfig(nodeNum, diff --git a/test/distributed/org/apache/cassandra/distributed/test/JVMDTestTest.java b/test/distributed/org/apache/cassandra/distributed/test/JVMDTestTest.java index 3a1a0a8d1e..7e5d99aa3f 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/JVMDTestTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/JVMDTestTest.java @@ -19,15 +19,22 @@ package org.apache.cassandra.distributed.test; import java.io.IOException; +import java.util.Collections; import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.TimeUnit; import java.util.logging.Logger; import org.junit.Assert; +import com.google.common.collect.ImmutableMap; import org.junit.Test; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.Feature; @@ -35,7 +42,11 @@ import org.apache.cassandra.distributed.api.LogAction; import org.apache.cassandra.service.CassandraDaemon; import org.apache.cassandra.utils.FBUtilities; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; public class JVMDTestTest extends TestBaseImpl @@ -66,7 +77,7 @@ public class JVMDTestTest extends TestBaseImpl try (Cluster cluster = init(Cluster.build(2).withConfig(c -> c.with(Feature.values())).start())) { // debug logging is turned on so we will see debug logs - Assert.assertFalse(cluster.get(1).logs().grep("^DEBUG").getResult().isEmpty()); + assertFalse(cluster.get(1).logs().grep("^DEBUG").getResult().isEmpty()); // make sure an exception is thrown in the cluster LogAction logs = cluster.get(2).logs(); long mark = logs.mark(); // get the current position so watching doesn't see any previous exceptions @@ -75,7 +86,83 @@ public class JVMDTestTest extends TestBaseImpl LoggerFactory.getLogger(CassandraDaemon.class).error("Error", new RuntimeException("fail without fail")); }); List errors = logs.watchFor(mark, "^ERROR").getResult(); - Assert.assertFalse(errors.isEmpty()); + assertFalse(errors.isEmpty()); + } + } + + @Test + public void nonSharedConfigClassTest() throws IOException + { + Map commitLogCompression = ImmutableMap.of("class_name", "org.apache.cassandra.io.compress.LZ4Compressor", + "parameters", Collections.emptyMap()); + Map encryptionOptions = ImmutableMap.of("cipher_suites", Collections.singletonList("FakeCipher"), + "optional", false, + "enabled", false); + + try (Cluster cluster = Cluster.build(1) + .withConfig(c -> { + c.set("concurrent_reads", 321); + c.set("internode_compression", Config.InternodeCompression.dc); + c.set("client_encryption_options", encryptionOptions); + c.set("commitlog_compression", commitLogCompression); + }).start()) + { + cluster.get(1).runOnInstance(() -> { + assertEquals(321, DatabaseDescriptor.getConcurrentReaders()); + assertEquals(Config.InternodeCompression.dc, DatabaseDescriptor.internodeCompression()); + assertEquals("org.apache.cassandra.io.compress.LZ4Compressor", DatabaseDescriptor.getCommitLogCompression().class_name); + assertTrue(DatabaseDescriptor.getCommitLogCompression().parameters.isEmpty()); + }); + } + } + + @Test + public void modifySchemaWithStoppedNode() throws Throwable + { + try (Cluster cluster = init(Cluster.build().withNodes(2).withConfig(c -> c.with(Feature.GOSSIP).with(Feature.NETWORK)).start())) + { + assertFalse(cluster.get(1).isShutdown()); + assertFalse(cluster.get(2).isShutdown()); + cluster.schemaChangeIgnoringStoppedInstances("CREATE TABLE "+KEYSPACE+".tbl1 (id int primary key, i int)"); + + cluster.get(2).shutdown(true).get(1, TimeUnit.MINUTES); + assertFalse(cluster.get(1).isShutdown()); + assertTrue(cluster.get(2).isShutdown()); + cluster.schemaChangeIgnoringStoppedInstances("CREATE TABLE "+KEYSPACE+".tbl2 (id int primary key, i int)"); + + cluster.get(1).shutdown(true).get(1, TimeUnit.MINUTES); + assertTrue(cluster.get(1).isShutdown()); + assertTrue(cluster.get(2).isShutdown()); + + // both nodes down, nothing to record a schema change so should get an exception + Throwable thrown = null; + try + { + cluster.schemaChangeIgnoringStoppedInstances("CREATE TABLE " + KEYSPACE + ".tblX (id int primary key, i int)"); + } + catch (Throwable tr) + { + thrown = tr; + } + assertNotNull("Expected to fail with all nodes down", thrown); + + // Have to restart instance1 before instance2 as it is hard-coded as the seed in in-JVM configuration. + cluster.get(1).startup(); + cluster.get(2).startup(); + assertFalse(cluster.get(1).isShutdown()); + assertFalse(cluster.get(2).isShutdown()); + cluster.schemaChangeIgnoringStoppedInstances("CREATE TABLE "+KEYSPACE+".tbl3 (id int primary key, i int)"); + + assertRows(cluster.get(1).executeInternal("SELECT columnfamily_name FROM system.schema_columnfamilies WHERE keyspace_name = ?", KEYSPACE), + row("tbl1"), row("tbl2"), row("tbl3")); + assertRows(cluster.get(2).executeInternal("SELECT columnfamily_name FROM system.schema_columnfamilies WHERE keyspace_name = ?", KEYSPACE), + row("tbl1"), row("tbl2"), row("tbl3")); + + // Finally test schema can be changed with the first node down + cluster.get(1).shutdown(true).get(2, TimeUnit.MINUTES); + cluster.schemaChangeIgnoringStoppedInstances("CREATE TABLE "+KEYSPACE+".tbl4 (id int primary key, i int)"); + assertRows(cluster.get(2).executeInternal("SELECT columnfamily_name FROM system.schema_columnfamilies WHERE keyspace_name = ?", KEYSPACE), + row("tbl1"), row("tbl2"), row("tbl3"), row("tbl4")); } } } diff --git a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java new file mode 100644 index 0000000000..c132ffc93d --- /dev/null +++ b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java @@ -0,0 +1,54 @@ +/* + * 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.config; + +import java.util.Collections; +import java.util.Map; + +import com.google.common.collect.ImmutableMap; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertArrayEquals; + + +public class YamlConfigurationLoaderTest +{ + @Test + public void fromMapTest() + { + Integer storagePort = 123; + Config.CommitLogSync commitLogSync = Config.CommitLogSync.batch; + ParameterizedClass seedProvider = new ParameterizedClass("org.apache.cassandra.locator.SimpleSeedProvider", Collections.emptyMap()); + EncryptionOptions encryptionOptions = new EncryptionOptions.ClientEncryptionOptions(); + encryptionOptions.keystore = "myNewKeystore"; + encryptionOptions.cipher_suites = new String[] {"SomeCipher"}; + + Map map = ImmutableMap.of("storage_port", storagePort, + "commitlog_sync", commitLogSync, + "seed_provider", seedProvider, + "client_encryption_options", encryptionOptions); + Config config = YamlConfigurationLoader.fromMap(map, Config.class); + assertEquals(storagePort, config.storage_port); // Check a simple integer + assertEquals(commitLogSync, config.commitlog_sync); // Check an enum + assertEquals(seedProvider, config.seed_provider); // Check a parameterized class + assertEquals(encryptionOptions.keystore, config.client_encryption_options.keystore); // Check a nested object + assertArrayEquals(encryptionOptions.cipher_suites, config.client_encryption_options.cipher_suites); + } +}