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

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
This commit is contained in:
Jon Meredith 2020-10-22 10:54:55 -07:00 committed by David Capwell
parent 72941b9ec1
commit db5832120f
7 changed files with 267 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/java/org/apache/cassandra/config/ParameterizedClass.java
View File

@ -28,6 +28,11 @@ public class ParameterizedClass
public String class_name;
public Map<String, String> parameters;
public ParameterizedClass()
{
// for snakeyaml
}
public ParameterizedClass(String class_name, Map<String, String> parameters)
{
this.class_name = class_name;

63
src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
View File

@ -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> T fromMap(Map<String,Object> map, Class<T> 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<Object> createDefaultList(int initSize)
{
return Lists.newCopyOnWriteArrayList();
}
@Override
protected Map<Object, Object> createDefaultMap()
{
return Maps.newConcurrentMap();
}
@Override
protected Set<Object> createDefaultSet(int initSize)
{
return Sets.newConcurrentHashSet();
}
@Override
protected Set<Object> createDefaultSet()
{
return Sets.newConcurrentHashSet();
}
}
private static class MissingPropertiesChecker extends PropertyUtils
{
private final Set<String> missingProperties = new HashSet<>();

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

@ -330,7 +330,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
NetworkTopology topology = NetworkTopology.build(ipPrefix, broadcastPort, nodeIdTopology);
InstanceConfig config = InstanceConfig.generate(nodeNum, ipAddress, topology, root, String.valueOf(token), seedIp, datadirCount);
config.set("dtest.api.cluster_id", clusterId);
config.set("dtest.api.cluster_id", clusterId.toString());
if (configUpdater != null)
configUpdater.accept(config);
@ -387,6 +387,12 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
return instanceMap.get(addr);
}
public I getFirstRunningInstance()
{
return stream().filter(i -> !i.isShutdown()).findFirst().orElseThrow(
() -> new IllegalStateException("All instances are shutdown"));
}
public int size()
{
return instances.size();
@ -459,11 +465,31 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
public void schemaChange(String query)
{
get(1).sync(() -> {
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<I extends IInstance> implements ICluster<I
final SimpleCondition completed;
private final long timeOut;
private final TimeUnit timeoutUnit;
protected Predicate<IInstance> 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<I extends IInstance> implements ICluster<I
private void startPolling()
{
for (IInstance instance : instances)
cleanup.add(startPolling(instance));
instances.stream().filter(instanceFilter).forEach(instance -> cleanup.add(startPolling(instance)));
}
protected abstract IListen.Cancel startPolling(IInstance instance);
@ -575,7 +607,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
protected boolean isCompleted()
{
return 1 == instances.stream().map(IInstance::schemaVersion).distinct().count();
return 1 == instances.stream().filter(instanceFilter).map(IInstance::schemaVersion).distinct().count();
}
protected String getMonitorTimeoutMessage()

19
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -52,6 +52,7 @@ import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.config.YamlConfigurationLoader;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
@ -119,13 +120,6 @@ import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
public class Instance extends IsolatedExecutor implements IInvokableInstance
{
private static final Map<Class<?>, Function<Object, Object>> mapper = new HashMap<Class<?>, Function<Object, Object>>() {{
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<String,Object> params = ((InstanceConfig) overrides).getParams();
return YamlConfigurationLoader.fromMap(params, Config.class);
}
private void initializeRing(ICluster cluster)

49
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
View File

@ -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<Class<?>, Function<Object, Object>> mapping)
{
for (Map.Entry<String, Object> 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<Class<?>, Function<Object, Object>> 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<String, Object> 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,

91
test/distributed/org/apache/cassandra/distributed/test/JVMDTestTest.java
View File

@ -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<String> errors = logs.watchFor(mark, "^ERROR").getResult();
Assert.assertFalse(errors.isEmpty());
assertFalse(errors.isEmpty());
}
}
@Test
public void nonSharedConfigClassTest() throws IOException
{
Map<String,Object> commitLogCompression = ImmutableMap.of("class_name", "org.apache.cassandra.io.compress.LZ4Compressor",
"parameters", Collections.<String,Object>emptyMap());
Map<String,Object> 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"));
}
}
}

54
test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java Normal file
View File

@ -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<String,Object> 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);
}
}