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

Streaming exception race creates corrupt transaction log files that prevent restart 

patch by Jon Meredith; reviewed by Caleb Rackliffe, David Capwell for CASSANDRA-18736
This commit is contained in:
Jon Meredith 2024-04-25 13:53:28 -06:00
parent fba4a85b97
commit 9157d98e4c
8 changed files with 353 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.13
* Streaming exception race creates corrupt transaction log files that prevent restart (CASSANDRA-18736)
* Fix CQL tojson timestamp output on negative timestamp values before Gregorian calendar reform in 1582 (CASSANDRA-19566)
* Fix few types issues and implement types compatibility tests (CASSANDRA-19479)
* Optionally avoid hint transfer during decommission (CASSANDRA-19525)

36
src/java/org/apache/cassandra/db/lifecycle/LogFile.java
View File

@ -27,6 +27,8 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
@ -53,6 +55,7 @@ import static org.apache.cassandra.utils.Throwables.merge;
*
* @see LogTransaction
*/
@NotThreadSafe
final class LogFile implements AutoCloseable
{
private static final Logger logger = LoggerFactory.getLogger(LogFile.class);
@ -66,8 +69,9 @@ final class LogFile implements AutoCloseable
private final LogReplicaSet replicas = new LogReplicaSet();
// The transaction records, this set must be ORDER PRESERVING
private final Set<LogRecord> records = Collections.synchronizedSet(new LinkedHashSet<>()); // TODO: Hack until we fix CASSANDRA-14554
private final Set<LogRecord> onDiskRecords = Collections.synchronizedSet(new LinkedHashSet<>());
private final Set<LogRecord> records = new LinkedHashSet<>();
private final Set<LogRecord> onDiskRecords = new LinkedHashSet<>();
private boolean completed = false;
// The type of the transaction
private final OperationType type;
@ -123,6 +127,11 @@ final class LogFile implements AutoCloseable
deleteFilesForRecordsOfType(committed() ? Type.REMOVE : Type.ADD);
// safe to release memory for both types of records, the completed flag will prevent
// new records being added.
records.clear();
onDiskRecords.clear();
// we sync the parent directories between contents and log deletion
// to ensure there is a happens before edge between them
Throwables.maybeFail(syncDirectory(accumulate));
@ -162,6 +171,11 @@ final class LogFile implements AutoCloseable
logger.error("Failed to read records for transaction log {}", this);
return false;
}
LogRecord lastRecord = getLastRecord();
if (lastRecord != null &&
(lastRecord.type == Type.COMMIT || lastRecord.type == Type.ABORT) &&
lastRecord.isValid())
completed = true;
Set<String> absolutePaths = new HashSet<>();
for (LogRecord record : records)
@ -276,11 +290,13 @@ final class LogFile implements AutoCloseable
void commit()
{
addRecord(LogRecord.makeCommit(System.currentTimeMillis()));
completed = true;
}
void abort()
{
addRecord(LogRecord.makeAbort(System.currentTimeMillis()));
completed = true;
}
private boolean isLastRecordValidWithType(Type type)
@ -296,14 +312,9 @@ final class LogFile implements AutoCloseable
return isLastRecordValidWithType(Type.COMMIT);
}
boolean aborted()
{
return isLastRecordValidWithType(Type.ABORT);
}
boolean completed()
{
return committed() || aborted();
return completed;
}
void add(SSTable table)
@ -353,8 +364,8 @@ final class LogFile implements AutoCloseable
void addRecord(LogRecord record)
{
if (completed())
throw new IllegalStateException("Transaction already completed");
if (completed)
throw TransactionAlreadyCompletedException.create(getFiles());
if (records.contains(record))
throw new IllegalStateException("Record already exists");
@ -367,6 +378,9 @@ final class LogFile implements AutoCloseable
void remove(SSTable table)
{
if (completed)
throw TransactionAlreadyCompletedException.create(getFiles());
LogRecord record = makeAddRecord(table);
assert records.contains(record) : String.format("[%s] is not tracked by %s", record, id);
assert record.absolutePath.isPresent();
@ -401,8 +415,6 @@ final class LogFile implements AutoCloseable
for (List<File> toDelete : existingFiles.values())
LogFile.deleteRecordFiles(toDelete);
records.clear();
}
private static void deleteRecordFiles(List<File> existingFiles)

6
src/java/org/apache/cassandra/db/lifecycle/LogReplicaSet.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.db.lifecycle;
import java.io.File;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@ -28,6 +27,8 @@ import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -43,11 +44,12 @@ import org.apache.cassandra.utils.Throwables;
* @see LogReplica
* @see LogFile
*/
@NotThreadSafe
public class LogReplicaSet implements AutoCloseable
{
private static final Logger logger = LoggerFactory.getLogger(LogReplicaSet.class);
private final Map<File, LogReplica> replicasByFile = Collections.synchronizedMap(new LinkedHashMap<>()); // TODO: Hack until we fix CASSANDRA-14554
private final Map<File, LogReplica> replicasByFile = new LinkedHashMap<>();
private Collection<LogReplica> replicas()
{

36
src/java/org/apache/cassandra/db/lifecycle/TransactionAlreadyCompletedException.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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.db.lifecycle;
import java.io.File;
import java.util.List;
public class TransactionAlreadyCompletedException extends IllegalStateException
{
private TransactionAlreadyCompletedException(List<File> files)
{
super("Transaction already completed. " + files);
}
static TransactionAlreadyCompletedException create(List<File> files)
{
return new TransactionAlreadyCompletedException(files);
}
}

20
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

@ -36,6 +36,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.lifecycle.TransactionAlreadyCompletedException;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.*;
@ -560,6 +561,16 @@ public class StreamSession implements IEndpointStateChangeSubscriber
return state == State.COMPLETE;
}
/**
* Return if this session was failed or aborted
*
* @return true if session was failed or aborted
*/
public boolean isFailedOrAborted()
{
return state == State.FAILED || state == State.ABORTED;
}
public synchronized void messageReceived(StreamMessage message)
{
if (message.type != StreamMessage.Type.KEEP_ALIVE)
@ -650,6 +661,15 @@ public class StreamSession implements IEndpointStateChangeSubscriber
return closeSession(State.FAILED);
}
}
else if (e instanceof TransactionAlreadyCompletedException && isFailedOrAborted())
{
// StreamDeserializer threads may actively be writing SSTables when the stream
// is failed or canceled, which aborts the lifecycle transaction and throws an exception
// when any new SSTable is added. Since the stream has already failed, suppress
// extra streaming log failure messages.
logger.debug("Stream lifecycle transaction already completed after stream failure (ignore)", e);
return null;
}
logError(e);

208
test/distributed/org/apache/cassandra/distributed/test/streaming/StreamFailedWhileReceivingTest.java Normal file
View File

@ -0,0 +1,208 @@
/*
* 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.streaming;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.streaming.CassandraEntireSSTableStreamReader;
import org.apache.cassandra.db.streaming.CassandraIncomingFile;
import org.apache.cassandra.db.streaming.CassandraStreamManager;
import org.apache.cassandra.db.streaming.CassandraStreamReceiver;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.exceptions.StartupException;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.format.RangeAwareSSTableWriter;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.streaming.IncomingStream;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.messages.StreamMessageHeader;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
/** This is a somewhat brittle test to demonstrate transaction log corruption
when streaming is aborted as streamed sstables are added to the
transaction log concurrently.
The transaction log should not be modified after streaming
has aborted or completed it.
*/
public class StreamFailedWhileReceivingTest extends TestBaseImpl
{
@Test
public void zeroCopy() throws IOException
{
streamClose(true);
}
@Test
public void notZeroCopy() throws IOException
{
streamClose(false);
}
private void streamClose(boolean zeroCopyStreaming) throws IOException
{
try (Cluster cluster = Cluster.build(2)
.withInstanceInitializer(BBHelper::install)
.withConfig(c -> c.with(Feature.values())
.set("stream_entire_sstables", zeroCopyStreaming)
.set("autocompaction_on_startup_enabled", false))
.start())
{
init(cluster);
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int PRIMARY KEY)"));
IInvokableInstance node1 = cluster.get(1);
IInvokableInstance node2 = cluster.get(2);
for (int i = 1; i <= 100; i++)
node1.executeInternal(withKeyspace("INSERT INTO %s.tbl (pk) VALUES (?)"), i);
node1.flush(KEYSPACE);
// trigger streaming; expected to fail as streaming socket closed in the middle (currently this is an unrecoverable event)
node2.nodetoolResult("repair", "-full", KEYSPACE, "tbl");
node2.runOnInstance(() -> {
try
{
// use the startup logic to check for corrupt txn logfiles from the streaming failure
// quicker than restarting the instance to check
ColumnFamilyStore.scrubDataDirectories(Schema.instance.getTableMetadata(KEYSPACE, "tbl"));
}
catch (StartupException ex)
{
throw new RuntimeException(ex);
}
});
}
}
public static class BBHelper
{
static volatile StreamSession firstSession;
static CountDownLatch firstStreamAbort = new CountDownLatch(1); // per-instance
// CassandraStreamManager.prepareIncomingStream
@SuppressWarnings("unused")
public static IncomingStream prepareIncomingStream(StreamSession session, StreamMessageHeader header, @SuperCall Callable<IncomingStream> zuper) throws Exception
{
if (firstSession == null)
firstSession = session;
return zuper.call();
}
// CassandraStreamReceiver.abort
@SuppressWarnings("unused")
public static void abort(@SuperCall Callable<Integer> zuper) throws Exception
{
firstStreamAbort.countDown();
zuper.call();
}
// RangeAwareSSTableWriter.append
@SuppressWarnings("unused")
public static boolean append(UnfilteredRowIterator partition, @SuperCall Callable<Boolean> zuper) throws Exception
{
// handles compressed and non-compressed
if (isCaller(CassandraIncomingFile.class.getName(), "read"))
{
if (firstSession != null)
{
firstSession.abort();
// delay here until CassandraStreamReceiver abort is called on NonPeriodic tasks
firstStreamAbort.await(1, TimeUnit.MINUTES);
}
}
return zuper.call();
}
// ColumnFamilyStore.createWriter - for entire sstable streaming, before adding to LogTransaction
@SuppressWarnings("unused")
public static Descriptor newSSTableDescriptor(File directory, Version version, SSTableFormat.Type format, @SuperCall Callable<Descriptor> zuper) throws Exception
{
if (isCaller(CassandraEntireSSTableStreamReader.class.getName(), "read"))
// handles compressed and non-compressed
{
if (firstSession != null)
{
firstSession.abort();
// delay here until CassandraStreamReceiver abort is called on NonPeriodic tasks
firstStreamAbort.await(1, TimeUnit.MINUTES);
}
}
return zuper.call();
}
private static boolean isCaller(String klass, String method)
{
StackTraceElement[] stack = Thread.currentThread().getStackTrace();
for (int i = 0; i < stack.length; i++)
{
StackTraceElement e = stack[i];
if (klass.equals(e.getClassName()) && method.equals(e.getMethodName()))
return true;
}
return false;
}
public static void install(ClassLoader classLoader, Integer num)
{
if (num != 2) // only target the second instance
return;
new ByteBuddy().rebase(CassandraStreamManager.class)
.method(named("prepareIncomingStream").and(takesArguments(2)))
.intercept(MethodDelegation.to(BBHelper.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
new ByteBuddy().rebase(RangeAwareSSTableWriter.class)
.method(named("append").and(takesArguments(1)))
.intercept(MethodDelegation.to(BBHelper.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
new ByteBuddy().rebase(ColumnFamilyStore.class)
.method(named("newSSTableDescriptor").and(takesArguments(3)))
.intercept(MethodDelegation.to(BBHelper.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
new ByteBuddy().rebase(CassandraStreamReceiver.class)
.method(named("abort").and(takesArguments(0)))
.intercept(MethodDelegation.to(BBHelper.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
}
}
}

2
test/unit/org/apache/cassandra/Util.java
View File

@ -23,8 +23,10 @@ import java.io.Closeable;
import java.io.EOFException;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.Callable;

68
test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java
View File

@ -21,7 +21,9 @@ import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@ -32,6 +34,7 @@ import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
@ -45,15 +48,21 @@ import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.DiskOptimizationStrategy;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SpinningDiskOptimizationStrategy;
import org.apache.cassandra.schema.MockSchema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.AlwaysPresentFilter;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.AbstractTransactionalTest;
import org.apache.cassandra.utils.concurrent.Transactional;
@ -1359,6 +1368,28 @@ public class LogTransactionTest extends AbstractTransactionalTest
return listFiles(folder, Directories.FileType.FINAL);
}
// Used by listFiles - this test is deliberately racing with files being
// removed and cleaned up, so it is possible that files are removed between listing and getting their
// canonical names, in which case they can be dropped from the stream.
private static Stream<File> toCanonicalIgnoringNotFound(File file)
{
try
{
return Stream.of(file.getCanonicalFile());
}
catch (IOException io)
{
if (Throwables.isCausedBy(io, t -> t instanceof NoSuchFileException))
{
return Stream.empty();
}
else
{
throw new UncheckedIOException(io);
}
}
}
static Set<File> listFiles(File folder, Directories.FileType... types)
{
Collection<Directories.FileType> match = Arrays.asList(types);
@ -1366,16 +1397,33 @@ public class LogTransactionTest extends AbstractTransactionalTest
(file, type) -> match.contains(type),
Directories.OnTxnErr.IGNORE).list()
.stream()
.map(f -> {
try
{
return f.getCanonicalFile();
}
catch (IOException e)
{
throw new IOError(e);
}
})
.flatMap(LogTransactionTest::toCanonicalIgnoringNotFound)
.collect(Collectors.toSet());
}
final String DUMMY_KS = "ks";
final String DUMMY_TBL = "tbl";
final File dir = new File(".");
int nextGeneration = 1;
final Set<Component> dummyComponents = Collections.singleton(Component.DATA);
final TableMetadataRef dummyMetadata = TableMetadataRef.forOfflineTools(TableMetadata.minimal(DUMMY_KS, DUMMY_TBL));
final DiskOptimizationStrategy dummyOptimizationStrategy = new SpinningDiskOptimizationStrategy();
SSTable dummySSTable()
{
int id = nextGeneration;
nextGeneration++;
Descriptor descriptor = new Descriptor(dir, DUMMY_KS, DUMMY_TBL, id);
return new SSTable(descriptor, dummyComponents, dummyMetadata, dummyOptimizationStrategy)
{
};
}
@Test(expected = TransactionAlreadyCompletedException.class)
public void useAfterCompletedTest()
{
LogTransaction txnFile = new LogTransaction(OperationType.STREAM);
txnFile.abort(); // this should complete the txn
txnFile.trackNew(dummySSTable()); // expect an IllegalStateException here
}
}