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 cassandra-5.0

This commit is contained in:
Jon Meredith 2024-04-25 14:00:52 -06:00
parent 93692a4b17 858a2b2321
commit 6559983d6c
7 changed files with 333 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -39,6 +39,7 @@ Merged from 4.1:
* Memoize Cassandra verion and add a backoff interval for failed schema pulls (CASSANDRA-18902)
* Fix StackOverflowError on ALTER after many previous schema changes (CASSANDRA-19166)
Merged from 4.0:
* 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)
* Change logging to TRACE when failing to get peer certificate (CASSANDRA-19508)

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

@ -35,6 +35,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;
import org.apache.commons.lang3.StringUtils;
@ -65,6 +67,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);
@ -80,8 +83,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;
@ -139,6 +143,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));
@ -178,6 +187,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)
@ -292,11 +306,13 @@ final class LogFile implements AutoCloseable
void commit()
{
addRecord(LogRecord.makeCommit(currentTimeMillis()));
completed = true;
}
void abort()
{
addRecord(LogRecord.makeAbort(currentTimeMillis()));
completed = true;
}
private boolean isLastRecordValidWithType(Type type)
@ -312,14 +328,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)
@ -369,8 +380,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");
@ -383,6 +394,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();
@ -417,8 +431,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

@ -18,7 +18,6 @@
package org.apache.cassandra.db.lifecycle;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@ -27,6 +26,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.apache.cassandra.io.util.File;
import org.slf4j.Logger;
@ -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.util.List;
import org.apache.cassandra.io.util.File;
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);
}
}

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

@ -57,6 +57,7 @@ import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.CompactionStrategyManager;
import org.apache.cassandra.db.lifecycle.TransactionAlreadyCompletedException;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.File;
@ -617,6 +618,16 @@ public class StreamSession
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)
@ -706,12 +717,21 @@ public class StreamSession
return closeSession(State.FAILED, "Failed because there was an " + e.getClass().getCanonicalName() + " with state=" + state.name());
}
}
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);
if (channel.connected())
{
state(State.FAILED); // make sure subsequent error handling sees the session in a final state
state(State.FAILED); // make sure subsequent error handling sees the session in a final state
channel.sendControlMessage(new SessionFailedMessage()).awaitUninterruptibly();
}
StringBuilder failureReason = new StringBuilder("Failed because of an unknown exception\n");

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.IOException;
import java.util.concurrent.Callable;
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.RangeAwareSSTableWriter;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.File;
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 org.apache.cassandra.utils.concurrent.CountDownLatch;
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 = CountDownLatch.newCountDownLatch(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.decrement();
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.awaitUninterruptibly(1, TimeUnit.MINUTES);
}
}
return zuper.call();
}
// ColumnFamilyStore.newSSTableDescriptor - for entire sstable streaming, before adding to LogTransaction
@SuppressWarnings("unused")
public static Descriptor newSSTableDescriptor(File directory, Version version, @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.awaitUninterruptibly(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);
}
}
}

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

@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -47,9 +48,14 @@ import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Token;
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.SSTableId;
import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.sstable.format.big.BigFormat.Components;
@ -1440,4 +1446,36 @@ public class LogTransactionTest extends AbstractTransactionalTest
.flatMap(LogTransactionTest::toCanonicalIgnoringNotFound)
.collect(Collectors.toSet());
}
}
static final String DUMMY_KS = "ks";
static final String DUMMY_TBL = "tbl";
final File dir = new File(".");
Supplier<SequenceBasedSSTableId> idSupplier = SequenceBasedSSTableId.Builder.instance.generator(Stream.of());
final Set<Component> dummyComponents = Collections.singleton(SSTableFormat.Components.DATA);
SSTable dummySSTable()
{
SSTableId id = idSupplier.get();
Descriptor descriptor = new Descriptor(dir, DUMMY_KS, DUMMY_TBL, id);
SSTable.Builder<?, ?> builder = new SSTable.Builder<>(descriptor);
builder.setComponents(dummyComponents);
return new SSTable(builder, null)
{
@Override
public DecoratedKey getFirst() { return null; }
@Override
public DecoratedKey getLast() { return null; }
@Override
public AbstractBounds<Token> getBounds() { return null; }
};
}
@Test(expected = TransactionAlreadyCompletedException.class)
public void useAfterCompletedTest()
{
try (LogTransaction txnFile = new LogTransaction(OperationType.STREAM))
{
txnFile.abort(); // this should complete the txn
txnFile.trackNew(dummySSTable()); // expect an IllegalStateException here
}
}}