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

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Brandon Williams 2023-03-09 10:04:52 -06:00
parent f14fa978e4 0ea1d384d4
commit bace51ce8e
6 changed files with 320 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -140,6 +140,7 @@ Merged from 4.0:
* Restore internode custom tracing on 4.0's new messaging system (CASSANDRA-17981)
Merged from 3.11:
Merged from 3.0:
* Fix default file system error handler for disk_failure_policy die (CASSANDRA-18294)
* Introduce check for names of test classes (CASSANDRA-17964)
* Suppress CVE-2022-41915 (CASSANDRA-18147)
* Suppress CVE-2021-1471, CVE-2021-3064, CVE-2021-4235 (CASSANDRA-18149)

2
src/java/org/apache/cassandra/service/DefaultFSErrorHandler.java
View File

@ -42,6 +42,7 @@ public class DefaultFSErrorHandler implements FSErrorHandler
switch (DatabaseDescriptor.getDiskFailurePolicy())
{
case die:
case stop_paranoid:
// exception not logged here on purpose as it is already logged
logger.error("Stopping transports as disk_failure_policy is " + DatabaseDescriptor.getDiskFailurePolicy());
@ -58,6 +59,7 @@ public class DefaultFSErrorHandler implements FSErrorHandler
switch (DatabaseDescriptor.getDiskFailurePolicy())
{
case die:
case stop_paranoid:
case stop:
// exception not logged here on purpose as it is already logged

71
test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorCorruptSSTableExceptionTest.java → test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorThrowableTest.java
View File

@ -19,11 +19,15 @@
package org.apache.cassandra.distributed.test;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.Config.DiskFailurePolicy;
@ -40,6 +44,8 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableCallab
import org.apache.cassandra.distributed.shared.AbstractBuilder;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
import org.apache.cassandra.io.sstable.format.ForwardingSSTableReader;
@ -52,26 +58,52 @@ import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseImpl
@RunWith(Parameterized.class)
public class JVMStabilityInspectorThrowableTest extends TestBaseImpl
{
@Test
public void testAbstractLocalAwareExecutorPlusOnIgnoredDiskFailurePolicy() throws Exception
private DiskFailurePolicy testPolicy;
private boolean testCorrupted;
private boolean expectNativeTransportRunning;;
private boolean expectGossiperEnabled;
public JVMStabilityInspectorThrowableTest(DiskFailurePolicy policy, boolean testCorrupted,
boolean expectNativeTransportRunning, boolean expectGossiperEnabled)
{
test(DiskFailurePolicy.ignore, true, true);
this.testPolicy = policy;
this.testCorrupted = testCorrupted;
this.expectNativeTransportRunning = expectNativeTransportRunning;
this.expectGossiperEnabled = expectGossiperEnabled;
}
@Parameterized.Parameters
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{ DiskFailurePolicy.ignore, true, true, true},
{ DiskFailurePolicy.stop, true, true, true},
{ DiskFailurePolicy.stop_paranoid, true, false, false},
{ DiskFailurePolicy.best_effort, true, true, true},
{ DiskFailurePolicy.ignore, false, true, true},
{ DiskFailurePolicy.stop, false, false, false},
{ DiskFailurePolicy.stop_paranoid, false, false, false},
{ DiskFailurePolicy.best_effort, false, true, true}
}
);
}
@Test
public void testAbstractLocalAwareExecutorPlusOnStopParanoidDiskFailurePolicy() throws Exception
public void testAbstractLocalAwareExecutorServiceOnPolicies() throws Exception
{
test(DiskFailurePolicy.stop_paranoid, false, false);
test(testPolicy, testCorrupted, expectNativeTransportRunning, expectGossiperEnabled);
}
private static void test(DiskFailurePolicy policy, boolean expectNativeTransportRunning, boolean expectGossiperEnabled) throws Exception
private static void test(DiskFailurePolicy policy, boolean shouldTestCorrupted, boolean expectNativeTransportRunning, boolean expectGossiperEnabled) throws Exception
{
String table = policy.name();
try (final Cluster cluster = init(getCluster(policy).start()))
{
cluster.setUncaughtExceptionsFilter(t -> Throwables.anyCauseMatches(t, t2 -> t2.getClass().getCanonicalName().equals(CorruptSSTableException.class.getCanonicalName())));
cluster.setUncaughtExceptionsFilter(t -> Throwables.anyCauseMatches(
t, t2 -> Arrays.asList(CorruptSSTableException.class.getCanonicalName(), FSReadError.class.getCanonicalName()).contains(t2.getClass().getCanonicalName())));
IInvokableInstance node = cluster.get(1);
boolean[] setup = node.callOnInstance(() -> {
CassandraDaemon instanceForTesting = CassandraDaemon.getInstanceForTesting();
@ -86,16 +118,16 @@ public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseIm
cluster.schemaChange("CREATE TABLE " + KEYSPACE + "." + table + " (id bigint PRIMARY KEY)");
node.executeInternal("INSERT INTO " + KEYSPACE + "." + table + " (id) VALUES (?)", 0L);
corruptTable(node, KEYSPACE, table);
throwThrowable(node, KEYSPACE, table, shouldTestCorrupted);
try
{
cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + '.' + table + " WHERE id=?", ConsistencyLevel.ONE, 0L);
Assert.fail("Select should fail as we corrupted SSTable on purpose.");
Assert.fail("Select should fail as we expect corrupted sstable or FS error.");
}
catch (final Exception ex)
{
// we expect that above query fails as we corrupted an sstable
// we expect that above query fails as we corrupted an sstable or throw FS error when read
}
waitForStop(!expectGossiperEnabled, node, new SerializableCallable<Boolean>()
@ -156,7 +188,7 @@ public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseIm
}
}
private static void corruptTable(IInvokableInstance node, String keyspace, String table)
private static void throwThrowable(IInvokableInstance node, String keyspace, String table, boolean shouldTestCorrupted)
{
node.runOnInstance(() -> {
ColumnFamilyStore cf = Keyspace.open(keyspace).getColumnFamilyStore(table);
@ -165,7 +197,7 @@ public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseIm
Set<SSTableReader> remove = cf.getLiveSSTables();
Set<SSTableReader> replace = new HashSet<>();
for (SSTableReader r : remove)
replace.add(new CorruptedSSTableReader(r));
replace.add(new CorruptedSSTableReader(r, shouldTestCorrupted));
cf.getTracker().removeUnsafe(remove);
cf.addSSTables(replace);
@ -182,20 +214,29 @@ public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseIm
private static final class CorruptedSSTableReader extends ForwardingSSTableReader
{
public CorruptedSSTableReader(SSTableReader delegate)
private boolean shouldThrowCorrupted;
public CorruptedSSTableReader(SSTableReader delegate, boolean shouldThrowCorrupted)
{
super(delegate);
this.shouldThrowCorrupted = shouldThrowCorrupted;
}
@Override
public UnfilteredRowIterator rowIterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, SSTableReadsListener listener)
{
throw throwCorrupted();
if (shouldThrowCorrupted)
throw throwCorrupted();
throw throwFSError();
}
private CorruptSSTableException throwCorrupted()
{
throw new CorruptSSTableException(new IOException("failed to get position"), descriptor.baseFile());
}
private FSError throwFSError()
{
throw new FSReadError(new IOException("failed to get position"), descriptor.baseFile());
}
}
}

121
test/unit/org/apache/cassandra/service/DefaultFSErrorHandlerTest.java Normal file
View File

@ -0,0 +1,121 @@
/*
* 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.service;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.FSErrorHandler;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@RunWith(Parameterized.class)
public class DefaultFSErrorHandlerTest
{
private FSErrorHandler handler = new DefaultFSErrorHandler();
Config.DiskFailurePolicy oldDiskPolicy;
Config.DiskFailurePolicy testDiskPolicy;
private boolean gossipRunningFSError;
private boolean gossipRunningCorruptedSStableException;
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
SchemaLoader.prepareServer();
CassandraDaemon daemon = new CassandraDaemon();
daemon.completeSetup(); //startup must be completed, otherwise FS error will kill JVM regardless of failure policy
StorageService.instance.registerDaemon(daemon);
StorageService.instance.initServer();
}
@AfterClass
public static void shutdown()
{
StorageService.instance.stopClient();
}
@Before
public void setup()
{
StorageService.instance.startGossiping();
assertTrue(Gossiper.instance.isEnabled());
oldDiskPolicy = DatabaseDescriptor.getDiskFailurePolicy();
}
public DefaultFSErrorHandlerTest(Config.DiskFailurePolicy policy,
boolean gossipRunningFSError,
boolean gossipRunningCorruptedSStableException)
{
this.testDiskPolicy = policy;
this.gossipRunningFSError = gossipRunningFSError;
this.gossipRunningCorruptedSStableException = gossipRunningCorruptedSStableException;
}
@Parameterized.Parameters
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{ Config.DiskFailurePolicy.die, false, false},
{ Config.DiskFailurePolicy.ignore, true, true},
{ Config.DiskFailurePolicy.stop, false, true},
{ Config.DiskFailurePolicy.stop_paranoid, false, false},
{ Config.DiskFailurePolicy.best_effort, true, true}
}
);
}
@After
public void teardown()
{
DatabaseDescriptor.setDiskFailurePolicy(oldDiskPolicy);
}
@Test
public void testFSErrors()
{
DatabaseDescriptor.setDiskFailurePolicy(testDiskPolicy);
handler.handleFSError(new FSReadError(new IOException(), "blah"));
assertEquals(gossipRunningFSError, Gossiper.instance.isEnabled());
}
@Test
public void testCorruptSSTableException()
{
DatabaseDescriptor.setDiskFailurePolicy(testDiskPolicy);
handler.handleCorruptSSTable(new CorruptSSTableException(new IOException(), "blah"));
assertEquals(gossipRunningCorruptedSStableException, Gossiper.instance.isEnabled());
}
}

135
test/unit/org/apache/cassandra/service/DiskFailurePolicyTest.java Normal file
View File

@ -0,0 +1,135 @@
/*
* 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.service;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Config.DiskFailurePolicy;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.KillerForTests;
@RunWith(Parameterized.class)
public class DiskFailurePolicyTest
{
DiskFailurePolicy originalDiskFailurePolicy;
JVMStabilityInspector.Killer originalKiller;
KillerForTests killerForTests;
DiskFailurePolicy testPolicy;
boolean isStartUpInProgress;
Throwable t;
boolean expectGossipRunning;
boolean expectJVMKilled;
boolean expectJVMKilledQuiet;
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
SchemaLoader.prepareServer();
StorageService.instance.initServer();
FileUtils.setFSErrorHandler(new DefaultFSErrorHandler());
}
public DiskFailurePolicyTest(DiskFailurePolicy testPolicy, boolean isStartUpInProgress, Throwable t,
boolean expectGossipRunning, boolean jvmKilled, boolean jvmKilledQuiet)
{
this.testPolicy = testPolicy;
this.isStartUpInProgress = isStartUpInProgress;
this.t = t;
this.expectGossipRunning = expectGossipRunning;
this.expectJVMKilled = jvmKilled;
this.expectJVMKilledQuiet = jvmKilledQuiet;
}
@Parameterized.Parameters
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{ Config.DiskFailurePolicy.die, true, new FSReadError(new IOException(), "blah"), false, true, true},
{ DiskFailurePolicy.ignore, true, new FSReadError(new IOException(), "blah"), true, false, false},
{ DiskFailurePolicy.stop, true, new FSReadError(new IOException(), "blah"), false, true, true},
{ DiskFailurePolicy.stop_paranoid, true, new FSReadError(new IOException(), "blah"), false, true, true},
{ Config.DiskFailurePolicy.die, true, new CorruptSSTableException(new IOException(), "blah"), false, true, true},
{ DiskFailurePolicy.ignore, true, new CorruptSSTableException(new IOException(), "blah"), true, false, false},
{ DiskFailurePolicy.stop, true, new CorruptSSTableException(new IOException(), "blah"), false, true, true},
{ DiskFailurePolicy.stop_paranoid, true, new CorruptSSTableException(new IOException(), "blah"), false, true, true},
{ Config.DiskFailurePolicy.die, false, new FSReadError(new IOException(), "blah"), false, true, false},
{ DiskFailurePolicy.ignore, false, new FSReadError(new IOException(), "blah"), true, false, false},
{ DiskFailurePolicy.stop, false, new FSReadError(new IOException(), "blah"), false, false, false},
{ DiskFailurePolicy.stop_paranoid, false, new FSReadError(new IOException(), "blah"), false, false, false},
{ Config.DiskFailurePolicy.die, false, new CorruptSSTableException(new IOException(), "blah"), false, true, false},
{ DiskFailurePolicy.ignore, false, new CorruptSSTableException(new IOException(), "blah"), true, false, false},
{ DiskFailurePolicy.stop, false, new CorruptSSTableException(new IOException(), "blah"), true, false, false},
{ DiskFailurePolicy.stop_paranoid, false, new CorruptSSTableException(new IOException(), "blah"), false, false, false}
}
);
}
@Before
public void setup()
{
CassandraDaemon daemon = new CassandraDaemon();
if (!isStartUpInProgress)
daemon.completeSetup(); //mark startup completed
StorageService.instance.registerDaemon(daemon);
killerForTests = new KillerForTests();
originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
originalDiskFailurePolicy = DatabaseDescriptor.getDiskFailurePolicy();
StorageService.instance.startGossiping();
Assert.assertTrue(Gossiper.instance.isEnabled());
}
@After
public void teardown()
{
JVMStabilityInspector.replaceKiller(originalKiller);
DatabaseDescriptor.setDiskFailurePolicy(originalDiskFailurePolicy);
}
@Test
public void testPolicies()
{
DatabaseDescriptor.setDiskFailurePolicy(testPolicy);
JVMStabilityInspector.inspectThrowable(t);
Assert.assertEquals(expectJVMKilled, killerForTests.wasKilled());
Assert.assertEquals(expectJVMKilledQuiet, killerForTests.wasKilledQuietly());
if (!expectJVMKilled) {
// only verify gossip if JVM is not killed
Assert.assertEquals(expectGossipRunning, Gossiper.instance.isEnabled());
}
}
}

5
test/unit/org/apache/cassandra/utils/KillerForTests.java
View File

@ -45,6 +45,11 @@ public class KillerForTests extends JVMStabilityInspector.Killer
if (!expected)
Assert.fail("Saw JVM Kill but did not expect it.");
if (killed)
{
// Can only be killed once
return;
}
this.killed = true;
this.quiet = quiet;
}