diff --git a/CHANGES.txt b/CHANGES.txt index 1ea51847fb..1dd52c2dfd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,8 @@ * Avoid failing compactions with very large partitions (CASSANDRA-15164) * Use IF NOT EXISTS for index and UDT create statements in snapshot schema files (CASSANDRA-13935) * Add flag to ignore unreplicated keyspaces during repair (CASSANDRA-15160) +Merged from 2.2: + * Fixed a NullPointerException when calling nodetool enablethrift (CASSANDRA-16127) 3.0.22: * Fix gossip shutdown order (CASSANDRA-15816) diff --git a/build.xml b/build.xml index e3270d88ba..8ecf70b09c 100644 --- a/build.xml +++ b/build.xml @@ -399,6 +399,7 @@ + @@ -505,6 +506,7 @@ + @@ -528,6 +530,7 @@ version="${version}"/> + diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 666ca721a8..ce0b8f3396 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -28,16 +28,13 @@ import java.rmi.AlreadyBoundException; import java.rmi.NotBoundException; import java.rmi.Remote; import java.rmi.RemoteException; -import java.rmi.registry.LocateRegistry; import java.rmi.registry.Registry; import java.rmi.server.RMIClientSocketFactory; import java.rmi.server.RMIServerSocketFactory; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; - import javax.management.ObjectName; import javax.management.StandardMBean; import javax.management.remote.JMXConnectorServer; @@ -45,6 +42,13 @@ import javax.management.remote.JMXServiceURL; import javax.management.remote.rmi.RMIConnectorServer; import javax.management.remote.rmi.RMIJRMPServerImpl; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.Uninterruptibles; +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; @@ -53,20 +57,17 @@ 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 com.google.common.util.concurrent.Uninterruptibles; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.concurrent.*; +import org.apache.cassandra.batchlog.LegacyBatchlogMigrator; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; -import org.apache.cassandra.db.*; -import org.apache.cassandra.batchlog.LegacyBatchlogMigrator; +import org.apache.cassandra.cql3.functions.ThreadAwareSecurityManager; +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; @@ -79,10 +80,15 @@ 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.cql3.functions.ThreadAwareSecurityManager; import org.apache.cassandra.thrift.ThriftServer; import org.apache.cassandra.tracing.Tracing; -import org.apache.cassandra.utils.*; +import org.apache.cassandra.utils.FBUtilities; +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.RMIServerSocketFactoryImpl; +import org.apache.cassandra.utils.WindowsTimer; /** * The CassandraDaemon is an abstraction for a Cassandra daemon @@ -158,8 +164,8 @@ public class CassandraDaemon private static final CassandraDaemon instance = new CassandraDaemon(); - public Server thriftServer; - private NativeTransportService nativeTransportService; + private volatile Server thriftServer; + private volatile NativeTransportService nativeTransportService; private final boolean runManaged; protected final StartupChecks startupChecks; @@ -401,12 +407,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(); @@ -494,6 +500,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(); @@ -501,10 +509,15 @@ public class CassandraDaemon catch (IllegalStateException isx) { // If there are any errors, we just log and return in this case - logger.info(isx.getMessage()); + logger.warn(isx.getMessage()); return; } + startClientTransports(); + } + + private void startClientTransports() + { String nativeFlag = System.getProperty("cassandra.start_native_transport"); if ((nativeFlag != null && Boolean.parseBoolean(nativeFlag)) || (nativeFlag == null && DatabaseDescriptor.startNativeTransport())) { @@ -516,7 +529,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"); } @@ -531,10 +544,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 @@ -556,22 +566,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. @@ -622,6 +624,8 @@ public class CassandraDaemon } start(); + + logger.info("Startup complete"); } catch (Throwable e) { @@ -683,10 +687,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() @@ -694,19 +694,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() diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 69e337e7bb..06f034695e 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -389,7 +389,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } - 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() @@ -1338,7 +1337,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"); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 934b28d162..7f98cfc44c 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -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}. @@ -106,6 +109,13 @@ public abstract class AbstractCluster implements ICluster SHARED_CLASSES = findClassesMarkedForSharedClassLoader(); + private static final Predicate 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 +173,7 @@ public abstract class AbstractCluster implements ICluster)Instance::new, classLoader) @@ -718,5 +728,11 @@ public abstract class AbstractCluster implements ICluster findClassesMarkedForSharedClassLoader() + { + return new Reflections("org.apache.cassandra").getTypesAnnotatedWith(Shared.class).stream() + .map(Class::getName) + .collect(Collectors.toSet()); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 2c06066a66..3e7e365014 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -83,7 +83,6 @@ import org.apache.cassandra.io.sstable.IndexSummaryManager; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.io.util.DataOutputBuffer; -import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.net.CompactEndpointSerializationHelper; import org.apache.cassandra.net.IMessageSink; @@ -106,6 +105,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.NanoTimeToCurrentTimeMillis; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.Throwables; @@ -582,6 +582,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(); @@ -596,11 +597,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())) @@ -703,7 +705,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)) { @@ -733,7 +735,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), @@ -871,4 +873,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); + } + } + } } diff --git a/test/distributed/org/apache/cassandra/distributed/shared/Byteman.java b/test/distributed/org/apache/cassandra/distributed/shared/Byteman.java new file mode 100644 index 0000000000..bc27ec765e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/shared/Byteman.java @@ -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 klasses; + + public static Byteman createFromScripts(String... scripts) + { + List 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 scripts, List texts, Set 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 extractClasses(List texts) + { + Pattern pattern = Pattern.compile("^CLASS (.*)$"); + Set 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; + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/shared/Shared.java b/test/distributed/org/apache/cassandra/distributed/shared/Shared.java new file mode 100644 index 0000000000..a1047b6808 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/shared/Shared.java @@ -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 +{ +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java b/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java new file mode 100644 index 0000000000..3ac5028557 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java @@ -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 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 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> 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; + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/ClientNetworkStopStartTest.java b/test/distributed/org/apache/cassandra/distributed/test/ClientNetworkStopStartTest.java new file mode 100644 index 0000000000..1d23ac7f8e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ClientNetworkStopStartTest.java @@ -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 CASSANDRA-16127 + */ + @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 CASSANDRA-16127 + */ + @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 + { + 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 + { + 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"); + } + } +} diff --git a/test/resources/byteman/stream_failure.btm b/test/resources/byteman/stream_failure.btm new file mode 100644 index 0000000000..e40f7fe25e --- /dev/null +++ b/test/resources/byteman/stream_failure.btm @@ -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