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

update

This commit is contained in:
Alan Wang 2026-07-22 14:29:57 -07:00
parent 0030cf8cee
commit 4a4685e907
7 changed files with 98 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/java/org/apache/cassandra/db/SSTableImporter.java
View File

@ -247,8 +247,16 @@ public class SSTableImporter
}
catch (Throwable t)
{
logger.error("[{}] Failed adding SSTables", importID, t);
throw new RuntimeException("Failed adding SSTables", t);
if (isAccordEnabled)
{
String msg = "Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator";
throw new RuntimeException(msg, t);
}
else
{
logger.error("[{}] Failed adding SSTables", importID, t);
throw new RuntimeException("Failed adding SSTables", t);
}
}
logger.info("[{}] Done loading load new SSTables for {}/{}", importID, cfs.getKeyspaceName(), cfs.getTableName());

22
src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java
View File

@ -50,6 +50,7 @@ import org.apache.cassandra.db.streaming.CassandraOutgoingFile;
import org.apache.cassandra.dht.NormalizedRanges;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.InetAddressAndPort;
@ -96,6 +97,8 @@ public class CoordinatedTransfer
final Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext;
SingleTransferResult streamResult = SingleTransferResult.Init();
// If importTxnEpochMismatch is true, all replicas will deterministically not import the SSTables in the pending directory
volatile boolean importTxnEpochMismatch = false;
public CoordinatedTransfer(UUID importID, TableMetadata tableMetadata, Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext, long streamingEpoch, TokenRange allSSTableRanges)
{
@ -117,14 +120,27 @@ public class CoordinatedTransfer
LocalTransfers.instance().save(this);
stream();
performImportTxn();
LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
try
{
performImportTxn();
if (importTxnEpochMismatch)
{
LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
throw new RuntimeException("SSTable import failed because of a concurrent topology change; please retry the operation");
}
LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
}
catch (ReadTimeoutException e)
{
throw new RuntimeException("SSTable import failed locally; however the operation may still be applied by the recovery coordinator", e);
}
}
private void performImportTxn()
{
TableMetadatas tables = TableMetadatas.of(tableMetadata);
TxnRead read = TxnRead.createImport(tables, allSSTableRanges, streamResult.planId, streamingEpoch);
TxnRead read = TxnRead.createImport(tables, allSSTableRanges, importID, streamResult.planId, streamingEpoch);
TableMetadatasAndKeys tablesAndKeys = new TableMetadatasAndKeys(tables, read.keys());
Txn txn = new Txn.InMemory(Read, read.keys(), read, TxnQuery.NONE, null, tablesAndKeys);
IAccordService.IAccordResult<TxnResult> accordResult = AccordService.instance().coordinateAsync(tableMetadata.epoch.getEpoch(), txn, ConsistencyLevel.ALL, Dispatcher.RequestTime.forImmediateExecution());

6
src/java/org/apache/cassandra/service/accord/LocalTransfers.java
View File

@ -181,16 +181,14 @@ public class LocalTransfers
});
}
// This method will be called by every Accord command store executor of
// the ranges it intersects
public void activatePendingTransfers(TxnRead.ImportMetadata metadata)
public void activatePendingTransfers(TxnRead.ImportMetadata metadata, long executeAtEpoch)
{
lock.readLock().lock();
try
{
PendingLocalTransfer pendingLocalTransfer = local.get(metadata.getPlanId());
if (pendingLocalTransfer != null)
pendingLocalTransfer.activate();
pendingLocalTransfer.activate(metadata, executeAtEpoch);
}
finally
{

28
src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java
View File

@ -33,6 +33,7 @@ import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
@ -66,11 +67,31 @@ public class PendingLocalTransfer
/**
* Safely moves SSTables into the live set.
*/
public synchronized void activate()
public synchronized void activate(TxnRead.ImportMetadata metadata, long executeAtEpoch)
{
if (activated)
return;
CoordinatedTransfer coordinatedTransfer = LocalTransfers.instance.coordinating.get(metadata.getImportID());
boolean isCoordinator = coordinatedTransfer != null;
if (metadata.getStreamingEpoch() != executeAtEpoch)
{
logger.info("{} Failing activation of pending SSTables because streaming epoch {} != importTxn executeAt epoch {}",
logPrefix(), metadata.getStreamingEpoch(), executeAtEpoch);
if (isCoordinator)
{
coordinatedTransfer.importTxnEpochMismatch = true;
LocalTransfers.instance().scheduleCoordinatedTransferCleanup(coordinatedTransfer);
}
else
LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
activated = true;
return;
}
long startedActivation = currentTimeMillis();
logger.info("{} Activating transfer {}, {} ms since pending", logPrefix(), this, startedActivation - createdAt);
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(tableId);
@ -117,7 +138,10 @@ public class PendingLocalTransfer
long finishedActivation = currentTimeMillis();
logger.info("{} Finished activating transfer {} in {} ms", logPrefix(), this, finishedActivation - startedActivation);
LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
if (isCoordinator)
LocalTransfers.instance().scheduleCoordinatedTransferCleanup(coordinatedTransfer);
else
LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
}
@Override

4
src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
View File

@ -413,9 +413,7 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadC
public AsyncChain<Data> performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt)
{
Callable<Data> callable = () -> {
if (importMetadata.getStreamingEpoch() != executeAt.epoch())
throw new RuntimeException("SSTable import failed because of a concurrent topology change");
LocalTransfers.instance.activatePendingTransfers(importMetadata);
LocalTransfers.instance.activatePendingTransfers(importMetadata, executeAt.epoch());
return new TxnData();
};
return submit(executor, callable, callable);

24
src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
View File

@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@ -67,6 +68,7 @@ import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.UUIDSerializer;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.Invariants.require;
@ -204,9 +206,9 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
// the ImportTxn has custom logic to move SSTables in the pending directory to the live set. Because these txn's are performed
// using range reads, clients can see a window of inconsistency for read txn's that are concurrent with the ImportTxn. However,
// data will always be consistent.
public static TxnRead createImport(TableMetadatas tables, TokenRange range, TimeUUID planId, long streamingEpoch)
public static TxnRead createImport(TableMetadatas tables, TokenRange range, UUID importID, TimeUUID planId, long streamingEpoch)
{
ImportMetadata importMetadata = new ImportMetadata(planId, streamingEpoch);
ImportMetadata importMetadata = new ImportMetadata(importID, planId, streamingEpoch);
return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, null)), null, importMetadata);
}
@ -446,15 +448,22 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
// Metadata that is needed to be kept track of for ImportTxns
public static class ImportMetadata
{
private final UUID importID;
private final TimeUUID planId;
private final Long streamingEpoch;
public ImportMetadata(TimeUUID planId, Long streamingEpoch)
public ImportMetadata(UUID importID, TimeUUID planId, Long streamingEpoch)
{
this.importID = importID;
this.planId = planId;
this.streamingEpoch = streamingEpoch;
}
public UUID getImportID()
{
return importID;
}
public TimeUUID getPlanId()
{
return planId;
@ -473,7 +482,8 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
public boolean equals(ImportMetadata that)
{
return Objects.equals(this.planId, that.planId)
return Objects.equals(this.importID, that.importID)
&& Objects.equals(this.planId, that.planId)
&& (Objects.equals(this.streamingEpoch, that.streamingEpoch));
}
@ -488,12 +498,14 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
@Override
public void serialize(ImportMetadata importMetadata, DataOutputPlus out, Version version) throws IOException
{
UUIDSerializer.serializer.serialize(importMetadata.importID, out);
TimeUUID.Serializer.instance.serialize(importMetadata.planId, out);
out.writeLong(importMetadata.streamingEpoch);
}
public void skip(DataInputPlus in, Version version) throws IOException
{
UUIDSerializer.serializer.skip(in);
TimeUUID.Serializer.instance.skip(in);
in.readLong();
}
@ -501,15 +513,17 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
@Override
public ImportMetadata deserialize(DataInputPlus in, Version version) throws IOException
{
UUID importID = UUIDSerializer.serializer.deserialize(in);
TimeUUID planId = TimeUUID.Serializer.instance.deserialize(in);
Long streamingEpoch = in.readLong();
return new ImportMetadata(planId, streamingEpoch);
return new ImportMetadata(importID, planId, streamingEpoch);
}
@Override
public long serializedSize(ImportMetadata importMetadata, Version version)
{
long size = 0;
size += UUIDSerializer.serializer.serializedSize(importMetadata.importID);
size += TimeUUID.Serializer.instance.serializedSize(importMetadata.planId);
size += TypeSizes.LONG_SIZE;
return size;

30
test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java
View File

@ -140,10 +140,6 @@ public class AccordImportSSTableTest extends TestBaseImpl
}
}
/**
* This might have a potential issue with us throwing an exception instead of having a custom class to notify
* that the read was a failure.
*/
@Test
public void testSSTableImportWithConcurrentTopologyChangeFails() throws Throwable
{
@ -177,9 +173,14 @@ public class AccordImportSSTableTest extends TestBaseImpl
Set<String> paths = Set.of(file);
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true))
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("Failed adding SSTables");
.hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator")
.cause()
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("SSTable import failed because of a concurrent topology change; please retry the operation");
});
}, "importer");
importer.start();
cluster.get(1).runOnInstance(() -> {
@ -228,8 +229,7 @@ public class AccordImportSSTableTest extends TestBaseImpl
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
Set<String> paths = Set.of(file);
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true))
.isInstanceOf(RuntimeException.class)
.cause();
.isInstanceOf(RuntimeException.class);
});
Iterable<IInvokableInstance> up = cluster.stream()
@ -380,7 +380,13 @@ public class AccordImportSSTableTest extends TestBaseImpl
cluster.get(1).runOnInstance(() -> {
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
Set<String> paths = Set.of(file);
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true));
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true))
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator")
.cause()
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("SSTable import failed locally; however the operation may still be applied by the recovery coordinator");
});
// Wait until the recovery coordinator picks up the Import Txn
@ -452,7 +458,13 @@ public class AccordImportSSTableTest extends TestBaseImpl
cluster.get(1).runOnInstance(() -> {
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
Set<String> paths = Set.of(file);
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true));
Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true))
.isInstanceOf(RuntimeException.class)
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator")
.cause()
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("SSTable import failed locally; however the operation may still be applied by the recovery coordinator");
});
// Wait until the recovery coordinator picks up the Import Txn