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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
David Capwell 2020-10-12 11:05:34 -07:00
parent 925ad35a7d 42989cee28
commit 34dde962fc
12 changed files with 820 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -6,6 +6,8 @@
Merged from 3.0:
* Handle unexpected columns due to schema races (CASSANDRA-15899)
* Add flag to ignore unreplicated keyspaces during repair (CASSANDRA-15160)
Merged from 2.2:
* Fixed a NullPointerException when calling nodetool enablethrift (CASSANDRA-16127)
3.11.8
* Correctly interpret SASI's `max_compaction_flush_memory_in_mb` setting in megabytes not bytes (CASSANDRA-16071)

3
build.xml
View File

@ -425,6 +425,7 @@
<dependency groupId="junit" artifactId="junit" version="4.6" />
<dependency groupId="org.mockito" artifactId="mockito-core" version="3.2.4" />
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.5" />
<dependency groupId="org.reflections" artifactId="reflections" version="0.9.12" />
<dependency groupId="org.apache.rat" artifactId="apache-rat" version="0.10">
<exclusion groupId="commons-lang" artifactId="commons-lang"/>
</dependency>
@ -542,6 +543,7 @@
<dependency groupId="junit" artifactId="junit"/>
<dependency groupId="org.mockito" artifactId="mockito-core" />
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" />
<dependency groupId="org.reflections" artifactId="reflections" />
<dependency groupId="org.apache.rat" artifactId="apache-rat"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster"/>
@ -571,6 +573,7 @@
<dependency groupId="junit" artifactId="junit"/>
<dependency groupId="org.mockito" artifactId="mockito-core" />
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" />
<dependency groupId="org.reflections" artifactId="reflections" />
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded">
<exclusion groupId="io.netty" artifactId="netty-buffer"/>
<exclusion groupId="io.netty" artifactId="netty-codec"/>

102
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -26,11 +26,16 @@ import java.net.URL;
import java.net.UnknownHostException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.management.StandardMBean;
import javax.management.remote.JMXConnectorServer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.addthis.metrics3.reporter.config.ReporterConfig;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistryListener;
@ -39,12 +44,6 @@ import com.codahale.metrics.jvm.BufferPoolMetricSet;
import com.codahale.metrics.jvm.FileDescriptorRatioGauge;
import com.codahale.metrics.jvm.GarbageCollectorMetricSet;
import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.batchlog.LegacyBatchlogMigrator;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.CFMetaData;
@ -52,7 +51,11 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.config.SchemaConstants;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SizeEstimatesRecorder;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.WindowsFailedSnapshotTracker;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.StartupException;
@ -66,10 +69,16 @@ import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.metrics.DefaultNameFactory;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.schema.LegacySchemaMigrator;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.thrift.ThriftServer;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.Mx4jTool;
import org.apache.cassandra.utils.NativeLibrary;
import org.apache.cassandra.utils.WindowsTimer;
/**
* The <code>CassandraDaemon</code> is an abstraction for a Cassandra daemon
@ -159,8 +168,8 @@ public class CassandraDaemon
static final CassandraDaemon instance = new CassandraDaemon();
public Server thriftServer;
private NativeTransportService nativeTransportService;
private volatile Server thriftServer;
private volatile NativeTransportService nativeTransportService;
private JMXConnectorServer jmxServer;
private final boolean runManaged;
@ -426,12 +435,12 @@ public class CassandraDaemon
// due to scheduling errors or race conditions
ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(ColumnFamilyStore.getBackgroundCompactionTaskSubmitter(), 5, 1, TimeUnit.MINUTES);
initializeNativeTransport();
initializeClientTransports();
completeSetup();
}
public void initializeNativeTransport()
public synchronized void initializeClientTransports()
{
// Thrift
InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress();
@ -521,6 +530,8 @@ public class CassandraDaemon
*/
public void start()
{
// check to see if transports may start else return without starting. This is needed when in survey mode or
// when bootstrap has not completed.
try
{
validateTransportsCanStart();
@ -532,6 +543,11 @@ public class CassandraDaemon
return;
}
startClientTransports();
}
private void startClientTransports()
{
String nativeFlag = System.getProperty("cassandra.start_native_transport");
if ((nativeFlag != null && Boolean.parseBoolean(nativeFlag)) || (nativeFlag == null && DatabaseDescriptor.startNativeTransport()))
{
@ -543,7 +559,7 @@ public class CassandraDaemon
String rpcFlag = System.getProperty("cassandra.start_rpc");
if ((rpcFlag != null && Boolean.parseBoolean(rpcFlag)) || (rpcFlag == null && DatabaseDescriptor.startRpc()))
thriftServer.start();
startThriftServer();
else
logger.info("Not starting RPC server as requested. Use JMX (StorageService->startRPCServer()) or nodetool (enablethrift) to start it");
}
@ -558,10 +574,7 @@ public class CassandraDaemon
// On linux, this doesn't entirely shut down Cassandra, just the RPC server.
// jsvc takes care of taking the rest down
logger.info("Cassandra shutting down...");
if (thriftServer != null)
thriftServer.stop();
if (nativeTransportService != null)
nativeTransportService.destroy();
destroyClientTransports();
StorageService.instance.setRpcReady(false);
// On windows, we need to stop the entire system as prunsrv doesn't have the jsvc hooks
@ -583,22 +596,14 @@ public class CassandraDaemon
}
@VisibleForTesting
public void destroyNativeTransport() throws InterruptedException
public void destroyClientTransports()
{
stopThriftServer();
stopNativeTransport();
if (nativeTransportService != null)
{
nativeTransportService.destroy();
nativeTransportService = null;
}
if (thriftServer != null)
{
thriftServer.stop();
thriftServer = null;
}
}
/**
* Clean up all resources obtained during the lifetime of the daemon. This
* is a hook for JSVC.
@ -641,6 +646,8 @@ public class CassandraDaemon
}
start();
logger.info("Startup complete");
}
catch (Throwable e)
{
@ -707,10 +714,6 @@ public class CassandraDaemon
throw new IllegalStateException("setup() must be called first for CassandraDaemon");
nativeTransportService.start();
if (thriftServer == null)
throw new IllegalStateException("thrift transport should be set up before it can be started");
thriftServer.start();
}
public void stopNativeTransport()
@ -718,19 +721,34 @@ public class CassandraDaemon
if (nativeTransportService != null)
{
nativeTransportService.stop();
nativeTransportService = null;
}
if (thriftServer != null)
{
thriftServer.stop();
thriftServer = null;
}
}
public boolean isNativeTransportRunning()
{
return nativeTransportService != null ? nativeTransportService.isRunning() : false;
return nativeTransportService != null && nativeTransportService.isRunning();
}
public void startThriftServer()
{
validateTransportsCanStart();
if (thriftServer == null)
throw new IllegalStateException("setup() must be called first for CassandraDaemon");
thriftServer.start();
}
public void stopThriftServer()
{
if (thriftServer != null)
{
thriftServer.stop();
}
}
public boolean isThriftServerRunning()
{
return thriftServer != null && thriftServer.isRunning();
}
public int getMaxNativeProtocolVersion()

11
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -389,7 +389,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
throw new IllegalStateException("Not starting RPC server as write_survey mode and authentication is enabled");
}
daemon.thriftServer.start();
daemon.startThriftServer();
}
public void stopRPCServer()
@ -398,17 +398,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
throw new IllegalStateException("No configured daemon");
}
if (daemon.thriftServer != null)
daemon.thriftServer.stop();
daemon.stopThriftServer();
}
public boolean isRPCServerRunning()
{
if ((daemon == null) || (daemon.thriftServer == null))
if (daemon == null)
{
return false;
}
return daemon.thriftServer.isRunning();
return daemon.isThriftServerRunning();
}
public synchronized void startNativeTransport()
@ -1613,7 +1612,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
progressSupport.progress("bootstrap", new ProgressEvent(ProgressEventType.COMPLETE, 1, 1, "Resume bootstrap complete"));
if (!isNativeTransportRunning())
daemon.initializeNativeTransport();
daemon.initializeClientTransports();
daemon.start();
logger.info("Resume complete");
}

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

@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -65,12 +66,14 @@ import org.apache.cassandra.distributed.shared.AbstractBuilder;
import org.apache.cassandra.distributed.shared.InstanceClassLoader;
import org.apache.cassandra.distributed.shared.MessageFilters;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.shared.Shared;
import org.apache.cassandra.distributed.shared.ShutdownException;
import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.reflections.Reflections;
/**
* AbstractCluster creates, initializes and manages Cassandra instances ({@link Instance}.
@ -98,7 +101,8 @@ import org.apache.cassandra.utils.concurrent.SimpleCondition;
*/
public abstract class AbstractCluster<I extends IInstance> implements ICluster<I>, AutoCloseable
{
public static Versions.Version CURRENT_VERSION = new Versions.Version(FBUtilities.getReleaseVersionString(), Versions.getClassPath());;
public static Versions.Version CURRENT_VERSION = new Versions.Version(FBUtilities.getReleaseVersionString(), Versions.getClassPath());
// WARNING: we have this logger not (necessarily) for logging, but
// to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener)
@ -106,6 +110,13 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private static final Logger logger = LoggerFactory.getLogger(AbstractCluster.class);
private static final AtomicInteger GENERATION = new AtomicInteger();
// include byteman so tests can use
private static final Set<String> SHARED_CLASSES = findClassesMarkedForSharedClassLoader();
private static final Predicate<String> SHARED_PREDICATE = s ->
SHARED_CLASSES.contains(s) ||
InstanceClassLoader.getDefaultLoadSharedFilter().test(s) ||
s.startsWith("org.jboss.byteman");
private final UUID clusterId = UUID.randomUUID();
private final File root;
private final ClassLoader sharedClassLoader;
@ -163,7 +174,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private IInvokableInstance newInstance(int generation)
{
ClassLoader classLoader = new InstanceClassLoader(generation, config.num(), version.classpath, sharedClassLoader);
ClassLoader classLoader = new InstanceClassLoader(generation, config.num(), version.classpath, sharedClassLoader, SHARED_PREDICATE);
if (instanceInitializer != null)
instanceInitializer.accept(classLoader, config.num());
return Instance.transferAdhoc((SerializableBiFunction<IInstanceConfig, ClassLoader, IInvokableInstance>)Instance::new, classLoader)
@ -718,5 +729,11 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
.collect(Collectors.toList());
}
private static Set<String> findClassesMarkedForSharedClassLoader()
{
return new Reflections("org.apache.cassandra").getTypesAnnotatedWith(Shared.class).stream()
.map(Class::getName)
.collect(Collectors.toSet());
}
}

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

@ -106,6 +106,7 @@ import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.DiagnosticSnapshotService;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.concurrent.Ref;
@ -141,7 +142,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
// Set the config at instance creation, possibly before startup() has run on all other instances.
// setMessagingVersions below will call runOnInstance which will instantiate
// the MessagingService and dependencies preventing later changes to network parameters.
Config.setOverrideLoadConfig(() -> loadConfig(config));
Config single = loadConfig(config);
Config.setOverrideLoadConfig(() -> single);
}
@Override
@ -535,6 +537,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
}
// TODO: this is more than just gossip
StorageService.instance.registerDaemon(CassandraDaemon.getInstanceForTesting());
if (config.has(GOSSIP))
{
StorageService.instance.initServer();
@ -549,11 +552,12 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
SystemKeyspace.finishStartup();
CassandraDaemon.getInstanceForTesting().setupCompleted();
if (config.has(NATIVE_PROTOCOL))
{
CassandraDaemon.getInstanceForTesting().initializeNativeTransport();
CassandraDaemon.getInstanceForTesting().startNativeTransport();
StorageService.instance.setRpcReady(true);
CassandraDaemon.getInstanceForTesting().initializeClientTransports();
CassandraDaemon.getInstanceForTesting().start();
}
if (!FBUtilities.getBroadcastAddress().equals(broadcastAddress().getAddress()))
@ -657,7 +661,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
error = parallelRun(error, executor,
() -> StorageService.instance.setRpcReady(false),
CassandraDaemon.getInstanceForTesting()::destroyNativeTransport);
CassandraDaemon.getInstanceForTesting()::destroyClientTransports);
if (config.has(GOSSIP) || config.has(NETWORK))
{
@ -683,7 +687,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
);
error = parallelRun(error, executor,
() -> ScheduledExecutors.shutdownAndWait(1L, MINUTES),
MessagingService.instance()::shutdown
(IgnoreThrowingRunnable) MessagingService.instance()::shutdown
);
error = parallelRun(error, executor,
() -> StageManager.shutdownAndWait(1L, MINUTES),
@ -821,4 +825,23 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
}
return accumulate;
}
@FunctionalInterface
private interface IgnoreThrowingRunnable extends ThrowingRunnable
{
void doRun() throws Throwable;
@Override
default void run()
{
try
{
doRun();
}
catch (Throwable e)
{
JVMStabilityInspector.inspectThrowable(e);
}
}
}
}

207
test/distributed/org/apache/cassandra/distributed/shared/Byteman.java Normal file
View File

@ -0,0 +1,207 @@
/*
* 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.shared;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLStreamHandler;
import java.nio.charset.StandardCharsets;
import java.security.CodeSigner;
import java.security.CodeSource;
import java.security.ProtectionDomain;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.base.StandardSystemProperty;
import com.google.common.io.ByteStreams;
import com.google.common.io.Files;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jboss.byteman.agent.Transformer;
public final class Byteman
{
private static final Logger logger = LoggerFactory.getLogger(Byteman.class);
private static final boolean DEBUG_TRANSFORMATIONS = Boolean.getBoolean("cassandra.test.byteman.transformations.debug");
private static final Method METHOD;
private static final URL BYTEMAN;
static
{
try
{
Method method = ClassLoader.class.getDeclaredMethod("defineClass",
String.class, byte[].class, Integer.TYPE, Integer.TYPE,
ProtectionDomain.class);
method.setAccessible(true);
METHOD = method;
}
catch (NoSuchMethodException e)
{
throw new AssertionError(e);
}
try
{
// this is just to make it more clear when you inspect a class that it was created by byteman
// the code source will show it came from byteman:// which isn't a valid java URL (hence the stream handler
// override)
BYTEMAN = new URL(null, "byteman://", new URLStreamHandler() {
protected URLConnection openConnection(URL u)
{
throw new UnsupportedOperationException();
}
});
}
catch (MalformedURLException e)
{
throw new AssertionError(e);
}
}
private final Transformer transformer;
private final List<KlassDetails> klasses;
public static Byteman createFromScripts(String... scripts)
{
List<String> texts = Stream.of(scripts).map(p -> {
try
{
return Files.toString(new File(p), StandardCharsets.UTF_8);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}).collect(Collectors.toList());
return new Byteman(Arrays.asList(scripts), texts, extractClasses(texts));
}
public static Byteman createFromText(String text)
{
return new Byteman(Arrays.asList("invalid"), Arrays.asList(text), extractClasses(Arrays.asList(text)));
}
private Byteman(List<String> scripts, List<String> texts, Set<String> modifiedClassNames)
{
klasses = modifiedClassNames.stream().map(fullyQualifiedKlass -> {
try
{
Class<?> klass = Class.forName(fullyQualifiedKlass);
String klassPath = fullyQualifiedKlass.replace(".", "/");
byte[] bytes = ByteStreams.toByteArray(Thread.currentThread().getContextClassLoader().getResourceAsStream(klassPath + ".class"));
return new KlassDetails(klassPath, klass, klass.getProtectionDomain(), bytes);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}).collect(Collectors.toList());
try
{
this.transformer = new Transformer(null, null, scripts, texts, false);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
public void install(ClassLoader cl)
{
try
{
for (KlassDetails details : klasses)
{
byte[] newBytes = transformer.transform(cl, details.klassPath, details.klass, details.protectionDomain, details.bytes);
if (newBytes == null)
throw new AssertionError("Unable to transform bytes for " + details.klassPath);
// inject the bytes into the classloader
METHOD.invoke(cl, null, newBytes, 0, newBytes.length,
new ProtectionDomain(new CodeSource(BYTEMAN, new CodeSigner[0]), details.protectionDomain.getPermissions()));
if (DEBUG_TRANSFORMATIONS)
{
File f = new File(StandardSystemProperty.JAVA_IO_TMPDIR.value(), "byteman/" + details.klassPath + ".class");
f.getParentFile().mkdirs();
File original = new File(f.getParentFile(), "original-" + f.getName());
logger.info("Writing class file for {} to {}", details.klassPath, f.getAbsolutePath());
Files.asByteSink(f).write(newBytes);
Files.asByteSink(original).write(details.bytes);
}
}
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
private static Set<String> extractClasses(List<String> texts)
{
Pattern pattern = Pattern.compile("^CLASS (.*)$");
Set<String> modifiedClassNames = new HashSet<>();
for (String text : texts)
{
for (String line : text.split("\n"))
{
Matcher matcher = pattern.matcher(line);
if (!matcher.find())
continue;
modifiedClassNames.add(matcher.group(1));
}
}
if (modifiedClassNames.isEmpty())
throw new AssertionError("Unable to find any classes to modify");
return modifiedClassNames;
}
private static final class KlassDetails
{
private final String klassPath;
private final Class<?> klass;
private final ProtectionDomain protectionDomain;
private final byte[] bytes;
public KlassDetails(String klassPath,
Class<?> klass, ProtectionDomain protectionDomain, byte[] bytes)
{
this.klassPath = klassPath;
this.klass = klass;
this.protectionDomain = protectionDomain;
this.bytes = bytes;
}
}
}

37
test/distributed/org/apache/cassandra/distributed/shared/Shared.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.shared;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Tells jvm-dtest that a class should be shared accross all {@link ClassLoader}s.
*
* Jvm-dtest relies on classloader isolation to run multiple cassandra instances in the same JVM, this makes it
* so some classes do not get shared (outside a blesssed set of classes/packages). When the default behavior
* is not desirable, this annotation will tell jvm-dtest to share the class accross all class loaders.
*/
@Retention(RetentionPolicy.RUNTIME)
@Target({ ElementType.TYPE })
public @interface Shared
{
}

165
test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java Normal file
View File

@ -0,0 +1,165 @@
/*
* 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.test;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.LogResult;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.Byteman;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.shared.Shared;
/**
* Replaces python dtest bootstrap_test.py::TestBootstrap::test_bootstrap_binary_disabled
*/
public class BootstrapBinaryDisabledTest extends TestBaseImpl
{
@Test
public void test() throws IOException, TimeoutException
{
Map<String, Object> config = new HashMap<>();
config.put("authenticator", "org.apache.cassandra.auth.PasswordAuthenticator");
config.put("authorizer", "org.apache.cassandra.auth.CassandraAuthorizer");
config.put("role_manager", "org.apache.cassandra.auth.CassandraRoleManager");
config.put("permissions_validity_in_ms", 0);
config.put("roles_validity_in_ms", 0);
int originalNodeCount = 1;
int expandedNodeCount = originalNodeCount + 2;
Byteman byteman = Byteman.createFromScripts("test/resources/byteman/stream_failure.btm");
try (Cluster cluster = init(Cluster.build(originalNodeCount)
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(expandedNodeCount))
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0"))
.withConfig(c -> {
config.forEach(c::set);
c.with(Feature.GOSSIP, Feature.NETWORK, Feature.NATIVE_PROTOCOL);
})
.withInstanceInitializer((cl, nodeNumber) -> {
switch (nodeNumber) {
case 1:
case 2:
byteman.install(cl);
break;
}
})
.start()))
{
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk text primary key)");
populate(cluster.get(1));
cluster.forEach(c -> c.flush(KEYSPACE));
bootstrap(cluster, config, false);
// Test write survey behaviour
bootstrap(cluster, config, true);
}
}
private static void bootstrap(Cluster cluster,
Map<String, Object> config,
boolean isWriteSurvey) throws TimeoutException
{
IInstanceConfig nodeConfig = cluster.newInstanceConfig();
nodeConfig.set("auto_bootstrap", true);
config.forEach(nodeConfig::set);
//TODO can we make this more isolated?
System.setProperty("cassandra.ring_delay_ms", "5000");
if (isWriteSurvey)
System.setProperty("cassandra.write_survey", "true");
RewriteEnabled.enable();
cluster.bootstrap(nodeConfig).startup();
IInvokableInstance node = cluster.get(cluster.size());
assertLogHas(node, "Some data streaming failed");
assertLogHas(node, isWriteSurvey ?
"Not starting client transports in write_survey mode as it's bootstrapping or auth is enabled" :
"Node is not yet bootstrapped completely");
node.nodetoolResult("join").asserts()
.failure()
.errorContains("Cannot join the ring until bootstrap completes");
RewriteEnabled.disable();
node.nodetoolResult("bootstrap", "resume").asserts().success();
if (isWriteSurvey)
assertLogHas(node, "Not starting client transports in write_survey mode as it's bootstrapping or auth is enabled");
if (isWriteSurvey)
{
node.nodetoolResult("join").asserts().success();
assertLogHas(node, "Leaving write survey mode and joining ring at operator request");
}
node.logs().watchFor("Starting listening for CQL clients");
assertBootstrapState(node, "COMPLETED");
}
private static void assertBootstrapState(IInvokableInstance node, String expected)
{
SimpleQueryResult qr = node.executeInternalWithResult("SELECT bootstrapped FROM system.local WHERE key='local'");
Assert.assertTrue("No rows found", qr.hasNext());
Assert.assertEquals(expected, qr.next().getString("bootstrapped"));
}
private static void assertLogHas(IInvokableInstance node, String msg)
{
LogResult<List<String>> results = node.logs().grep(msg);
Assert.assertFalse("Unable to find '" + msg + "'", results.getResult().isEmpty());
}
private void populate(IInvokableInstance inst)
{
for (int i = 0; i < 10; i++)
inst.executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)", Integer.toString(i));
}
@Shared
public static final class RewriteEnabled
{
private static volatile boolean enabled = false;
public static boolean isEnabled()
{
return enabled;
}
public static void enable()
{
enabled = true;
}
public static void disable()
{
enabled = false;
}
}
}

87
test/distributed/org/apache/cassandra/distributed/test/BytemanExamples.java Normal file
View File

@ -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.test;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import com.google.common.io.Files;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.shared.Byteman;
public class BytemanExamples
{
@Test
public void rewriteFromText() throws IOException
{
Byteman byteman = Byteman.createFromText("RULE example\n" +
"CLASS org.apache.cassandra.utils.FBUtilities\n" +
"METHOD setBroadcastInetAddress\n" +
"IF true\n" +
"DO\n" +
" throw new java.lang.RuntimeException(\"Will not allow init!\")\n" +
"ENDRULE\n");
try
{
Cluster.build(1)
.withInstanceInitializer((cl, ignore) -> byteman.install(cl))
.createWithoutStarting();
Assert.fail("Instance init was rewritten to fail right away, so make sure the rewrite happens");
}
catch (RuntimeException e)
{
Assert.assertEquals("Will not allow init!", e.getMessage());
}
}
@Test
public void rewriteFromScript() throws IOException
{
// scripts are normally located at test/resources/byteman, but generated in the test
// for documentation purposes only
File script = File.createTempFile("byteman", ".btm");
script.deleteOnExit();
Files.asCharSink(script, StandardCharsets.UTF_8).write("RULE example\n" +
"CLASS org.apache.cassandra.utils.FBUtilities\n" +
"METHOD setBroadcastInetAddress\n" +
"IF true\n" +
"DO\n" +
" throw new java.lang.RuntimeException(\"Will not allow init!\")\n" +
"ENDRULE\n");
Byteman byteman = Byteman.createFromScripts(script.getAbsolutePath());
try
{
Cluster.build(1)
.withInstanceInitializer((cl, ignore) -> byteman.install(cl))
.createWithoutStarting();
Assert.fail("Instance init was rewritten to fail right away, so make sure the rewrite happens");
}
catch (RuntimeException e)
{
Assert.assertEquals("Will not allow init!", e.getMessage());
}
}
}

192
test/distributed/org/apache/cassandra/distributed/test/ClientNetworkStopStartTest.java Normal file
View File

@ -0,0 +1,192 @@
/*
* 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.test;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.util.Arrays;
import java.util.Collections;
import java.util.Objects;
import org.junit.Assert;
import org.junit.Test;
import com.datastax.driver.core.Session;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.QueryResults;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TException;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
public class ClientNetworkStopStartTest extends TestBaseImpl
{
/**
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-16127">CASSANDRA-16127</a>
*/
@Test
public void stopStartThrift() throws IOException, TException
{
try (Cluster cluster = init(Cluster.build(1).withConfig(c -> c.with(Feature.NATIVE_PROTOCOL)).start()))
{
IInvokableInstance node = cluster.get(1);
assertTransportStatus(node, "binary", true);
assertTransportStatus(node, "thrift", true);
node.nodetoolResult("disablethrift").asserts().success();
assertTransportStatus(node, "binary", true);
assertTransportStatus(node, "thrift", false);
node.nodetoolResult("enablethrift").asserts().success();
assertTransportStatus(node, "binary", true);
assertTransportStatus(node, "thrift", true);
// now use it to make sure it still works!
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, value int, PRIMARY KEY (pk))");
ThriftClientUtils.thriftClient(node, thrift -> {
thrift.set_keyspace(KEYSPACE);
Mutation mutation = new Mutation();
ColumnOrSuperColumn csoc = new ColumnOrSuperColumn();
Column column = new Column();
column.setName(CompositeType.build(ByteBufferUtil.bytes("value")));
column.setValue(ByteBufferUtil.bytes(0));
column.setTimestamp(System.currentTimeMillis());
csoc.setColumn(column);
mutation.setColumn_or_supercolumn(csoc);
thrift.batch_mutate(Collections.singletonMap(ByteBufferUtil.bytes(0),
Collections.singletonMap("tbl", Arrays.asList(mutation))),
org.apache.cassandra.thrift.ConsistencyLevel.ALL);
});
SimpleQueryResult qr = cluster.coordinator(1).executeWithResult("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.ALL);
AssertUtils.assertRows(qr, QueryResults.builder().row(0, 0).build());
}
}
/**
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-16127">CASSANDRA-16127</a>
*/
@Test
public void stopStartNative() throws IOException
{
try (Cluster cluster = init(Cluster.build(1).withConfig(c -> c.with(Feature.NATIVE_PROTOCOL)).start()))
{
IInvokableInstance node = cluster.get(1);
assertTransportStatus(node, "binary", true);
assertTransportStatus(node, "thrift", true);
node.nodetoolResult("disablebinary").asserts().success();
assertTransportStatus(node, "binary", false);
assertTransportStatus(node, "thrift", true);
node.nodetoolResult("enablebinary").asserts().success();
assertTransportStatus(node, "binary", true);
assertTransportStatus(node, "thrift", true);
// now use it to make sure it still works!
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, value int, PRIMARY KEY (pk))");
try (com.datastax.driver.core.Cluster client = com.datastax.driver.core.Cluster.builder().addContactPoints(node.broadcastAddress().getAddress()).build();
Session session = client.connect())
{
session.execute("INSERT INTO " + KEYSPACE + ".tbl (pk, value) VALUES (?, ?)", 0, 0);
}
SimpleQueryResult qr = cluster.coordinator(1).executeWithResult("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.ALL);
AssertUtils.assertRows(qr, QueryResults.builder().row(0, 0).build());
}
}
private static void assertTransportStatus(IInvokableInstance node, String transport, boolean running)
{
assertNodetoolStdout(node, running ? "running" : "not running", running ? "not running" : null, "status" + transport);
}
private static void assertNodetoolStdout(IInvokableInstance node, String expectedStatus, String notExpected, String... nodetool)
{
// without CASSANDRA-16057 need this hack
PrintStream previousStdout = System.out;
try
{
ByteArrayOutputStream out = new ByteArrayOutputStream();
PrintStream stdout = new PrintStream(out, true);
System.setOut(stdout);
node.nodetoolResult(nodetool).asserts().success();
stdout.flush();
String output = out.toString();
Assert.assertThat(output, new StringContains(expectedStatus));
if (notExpected != null)
Assert.assertThat(output, new StringNotContains(notExpected));
}
finally
{
System.setOut(previousStdout);
}
}
private static final class StringContains extends BaseMatcher<String>
{
private final String expected;
private StringContains(String expected)
{
this.expected = Objects.requireNonNull(expected);
}
public boolean matches(Object o)
{
return o.toString().contains(expected);
}
public void describeTo(Description description)
{
description.appendText("Expected to find '" + expected + "', but did not");
}
}
private static final class StringNotContains extends BaseMatcher<String>
{
private final String notExpected;
private StringNotContains(String expected)
{
this.notExpected = Objects.requireNonNull(expected);
}
public boolean matches(Object o)
{
return !o.toString().contains(notExpected);
}
public void describeTo(Description description)
{
description.appendText("Expected not to find '" + notExpected + "', but did");
}
}
}

14
test/resources/byteman/stream_failure.btm Normal file
View File

@ -0,0 +1,14 @@
#
# Inject streaming failure
#
# Before start streaming files in `StreamSession#prepare()` method,
# interrupt streaming by throwing RuntimeException.
#
RULE inject stream failure
CLASS org.apache.cassandra.streaming.StreamSession
METHOD prepare
AT INVOKE maybeCompleted
IF org.apache.cassandra.distributed.test.BootstrapBinaryDisabledTest$RewriteEnabled.isEnabled()
DO
throw new java.lang.RuntimeException("Triggering network failure")
ENDRULE