You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/05/29 11:52:48 UTC
(cassandra) branch trunk updated: Accord Fixes: - cfk pruning+prebootstrap=invalid future dependency - exclude retired ranges when filtering RX stillTouches - propagate uses incorrect lowEpoch when fetch finds additional owned/touched ranges - node.withEpoch should callback with TopologyRetiredException, not throw - Recovery can race with durable-applied pruning; must not send durable unless latest ballot on apply - removeRedundantDependencies was not slicing pre-bootstrap range calculation to participating ranges - [...]
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new b3f035be1e Accord Fixes: - cfk pruning+prebootstrap=invalid future dependency - exclude retired ranges when filtering RX stillTouches - propagate uses incorrect lowEpoch when fetch finds additional owned/touched ranges - node.withEpoch should callback with TopologyRetiredException, not throw - Recovery can race with durable-applied pruning; must not send durable unless latest ballot on apply - removeRedundantDependencies was not slicing pre-bootstrap range calculation to pa [...]
b3f035be1e is described below
commit b3f035be1ee2d0b08010a4ac773cec20a83ef606
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Fri Apr 11 11:40:18 2025 +0100
Accord Fixes:
- cfk pruning+prebootstrap=invalid future dependency
- exclude retired ranges when filtering RX stillTouches
- propagate uses incorrect lowEpoch when fetch finds additional owned/touched ranges
- node.withEpoch should callback with TopologyRetiredException, not throw
- Recovery can race with durable-applied pruning; must not send durable unless latest ballot on apply
- removeRedundantDependencies was not slicing pre-bootstrap range calculation to participating ranges
- NPE in TopologyManager.atLeast caused by referencing an epoch that has been GC'd
- use journal durableBeforePersister in burn test, not NOOP_PERSISTER
- ServerUtils.cleanupDirectory use tryDeleteRecursive
- FsyncRunnable shutdown
- fix NPE in AccordJournalBurnTest
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20688
---
modules/accord | 2 +-
src/java/org/apache/cassandra/io/util/File.java | 9 +++++
.../org/apache/cassandra/io/util/PathUtils.java | 46 +++++++++++++++++-----
src/java/org/apache/cassandra/journal/Flusher.java | 18 +++++++--
.../service/accord/AccordCommandStore.java | 4 +-
.../service/accord/AccordConfigurationService.java | 4 +-
.../accord/AccordJournalValueSerializers.java | 2 -
.../cassandra/service/accord/AccordKeyspace.java | 1 -
.../service/accord/AccordSafeCommandStore.java | 1 -
.../cassandra/service/accord/AccordService.java | 1 -
.../cassandra/service/accord/AccordTask.java | 5 +--
.../service/accord/CommandsForRanges.java | 2 +-
.../accord/interop/AccordInteropAdapter.java | 21 +++++-----
.../service/accord/interop/AccordInteropApply.java | 19 ++++-----
.../accord/interop/AccordInteropExecution.java | 6 ++-
.../accord/interop/AccordInteropPersist.java | 5 ++-
.../accord/serializers/ApplySerializers.java | 14 ++++---
.../accord/serializers/CheckStatusSerializers.java | 7 +++-
.../accord/serializers/ReadDataSerializers.java | 1 -
.../accord/serializers/ResultSerializers.java | 2 +-
.../accord/serializers/TxnRequestSerializer.java | 2 +-
.../service/accord/txn/AccordUpdateParameters.java | 2 -
.../cassandra/service/accord/txn/TxnNamedRead.java | 3 --
.../cassandra/service/accord/txn/TxnUpdate.java | 6 +--
.../distributed/test/HintsMaxSizeTest.java | 2 -
.../test/SSTableLoaderEncryptionOptionsTest.java | 2 -
.../test/log/FetchLogFromPeers2Test.java | 4 +-
.../distributed/test/tcm/SplitBrainTest.java | 2 -
.../service/accord/AccordJournalBurnTest.java | 23 ++++++++++-
.../unit/org/apache/cassandra/ServerTestUtils.java | 2 +-
.../commitlog/CommitLogSegmentManagerCDCTest.java | 2 -
.../db/commitlog/CommitlogShutdownTest.java | 2 -
.../cassandra/hints/HintServiceBytemanTest.java | 2 -
.../cassandra/io/util/SafeMemoryWriterTest.java | 2 -
.../serializers/CheckStatusSerializersTest.java | 4 +-
.../serializers/CommandsForKeySerializerTest.java | 1 -
.../cassandra/tcm/DiscoverySimulationTest.java | 2 -
37 files changed, 143 insertions(+), 90 deletions(-)
diff --git a/modules/accord b/modules/accord
index 7f95490b13..c5a984cfe4 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 7f95490b1390b7fc68a4ff4ced7f161bafd8776b
+Subproject commit c5a984cfe41bb8d8f1d7a4cb446c194f829a5dd1
diff --git a/src/java/org/apache/cassandra/io/util/File.java b/src/java/org/apache/cassandra/io/util/File.java
index de415388ed..e814e37b74 100644
--- a/src/java/org/apache/cassandra/io/util/File.java
+++ b/src/java/org/apache/cassandra/io/util/File.java
@@ -225,6 +225,15 @@ public class File implements Comparable<File>
PathUtils.deleteRecursive(toPathForWrite());
}
+ /**
+ * Deletes all files and subdirectories under "dir".
+ * @return false if the root cannot be deleted
+ */
+ public boolean tryDeleteRecursive()
+ {
+ return PathUtils.tryDeleteRecursive(toPathForWrite());
+ }
+
/**
* Try to delete the file on process exit.
*/
diff --git a/src/java/org/apache/cassandra/io/util/PathUtils.java b/src/java/org/apache/cassandra/io/util/PathUtils.java
index 8ddd939b4c..fa0a91543e 100644
--- a/src/java/org/apache/cassandra/io/util/PathUtils.java
+++ b/src/java/org/apache/cassandra/io/util/PathUtils.java
@@ -346,11 +346,22 @@ public final class PathUtils
private static void deleteRecursiveUsingNixCommand(Path path, boolean quietly)
{
String [] cmd = new String[]{ "rm", quietly ? "-rdf" : "-rd", path.toAbsolutePath().toString() };
+ IOException failure = null;
+ if (!quietly && !Files.exists(path))
+ failure = new NoSuchFileException(path.toString());
+
+ if (failure == null)
+ failure = tryDeleteRecursiveUsingNixCommand(path, quietly);
+
+ if (failure != null)
+ throw propagateUnchecked(failure, path, true);
+ }
+
+ private static IOException tryDeleteRecursiveUsingNixCommand(Path path, boolean quietly)
+ {
+ String[] cmd = new String[]{ "rm", quietly ? "-rdf" : "-rd", path.toAbsolutePath().toString() };
try
{
- if (!quietly && !Files.exists(path))
- throw new NoSuchFileException(path.toString());
-
Process p = Runtime.getRuntime().exec(cmd);
int result = p.waitFor();
@@ -363,24 +374,39 @@ public final class PathUtils
}
if (result != 0 && Files.exists(path))
- {
- logger.error("{} returned:\nstdout:\n{}\n\nstderr:\n{}", Arrays.toString(cmd), out, err);
- throw new IOException(String.format("%s returned non-zero exit code: %d%nstdout:%n%s%n%nstderr:%n%s", Arrays.toString(cmd), result, out, err));
- }
+ return new IOException(String.format("%s returned non-zero exit code: %d%nstdout:%n%s%n%nstderr:%n%s", Arrays.toString(cmd), result, out, err));
onDeletion.accept(path);
+ return null;
}
catch (IOException e)
{
- throw propagateUnchecked(e, path, true);
+ return e;
}
catch (InterruptedException e)
{
- Thread.currentThread().interrupt();
- throw new FSWriteError(e, path);
+ return new IOException("Interrupted while executing command " + Arrays.toString(cmd), e);
}
}
+
+ /**
+ * Deletes all files and subdirectories under "path".
+ * @param path file to be deleted
+ * @return false if the root cannot be deleted
+ */
+ public static boolean tryDeleteRecursive(Path path)
+ {
+ if (USE_NIX_RECURSIVE_DELETE.getBoolean() && path.getFileSystem() == java.nio.file.FileSystems.getDefault())
+ return null == tryDeleteRecursiveUsingNixCommand(path, true);
+
+ if (isDirectory(path))
+ forEach(path, PathUtils::tryDeleteRecursive);
+
+ // The directory should now be empty, so now it can be smoked
+ return tryDelete(path);
+ }
+
/**
* Deletes all files and subdirectories under "path".
* @param path file to be deleted
diff --git a/src/java/org/apache/cassandra/journal/Flusher.java b/src/java/org/apache/cassandra/journal/Flusher.java
index d48d80171b..02f85df0cb 100644
--- a/src/java/org/apache/cassandra/journal/Flusher.java
+++ b/src/java/org/apache/cassandra/journal/Flusher.java
@@ -143,10 +143,20 @@ final class Flusher<K, V>
}
}
+ private boolean hasWork()
+ {
+ return hasWork(fsyncStartedFor);
+ }
+
+ private boolean hasWork(long lastStartedAt)
+ {
+ return fsyncWaitingSince != lastStartedAt;
+ }
+
private void awaitWork() throws InterruptedException
{
long lastStartedAt = fsyncStartedFor;
- if (fsyncWaitingSince != lastStartedAt)
+ if (hasWork(lastStartedAt))
return;
awaitingWork = Thread.currentThread();
@@ -158,7 +168,7 @@ final class Flusher<K, V>
throw new InterruptedException();
}
- if (fsyncWaitingSince != lastStartedAt)
+ if (hasWork(lastStartedAt))
break;
LockSupport.park();
@@ -175,7 +185,9 @@ final class Flusher<K, V>
public void doRun(Interruptible.State state) throws InterruptedException
{
- awaitWork();
+ if (state == NORMAL) awaitWork();
+ else if (!hasWork()) return;
+
if (fsyncing == null)
fsyncing = journal.oldestActiveSegment();
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 4e8629696a..ab74a128a7 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -501,7 +501,9 @@ public class AccordCommandStore extends CommandStore
RedundantBefore redundantBefore;
if (safeRedundantBefore == null) redundantBefore = RedundantBefore.EMPTY;
else redundantBefore = safeRedundantBefore.redundantBefore;
- return new AccordCompactionInfo(id, redundantBefore, rangesForEpoch, tableId);
+ CommandStores.RangesForEpoch ranges = this.rangesForEpoch;
+ if (ranges == null) ranges = CommandStores.RangesForEpoch.EMPTY;
+ return new AccordCompactionInfo(id, redundantBefore, ranges, tableId);
}
public RangeSearcher rangeSearcher()
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
index e0ca26ee5d..fade204b1f 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
@@ -38,6 +38,7 @@ import accord.primitives.Ranges;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.utils.Invariants;
+import accord.utils.SortedListSet;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.agrona.collections.LongArrayList;
@@ -449,8 +450,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
epochState.setSyncStatus(SyncStatus.NOTIFYING);
}
- // TODO (required): replace with SortedArraySet when it is available
- Set<Node.Id> notify = new HashSet<>(topology.nodes());
+ Set<Node.Id> notify = SortedListSet.allOf(topology.nodes());
notify.remove(localId);
syncPropagator.reportSyncComplete(epoch, notify, localId);
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
index e72beaab2c..58b238d31f 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
@@ -240,8 +240,6 @@ public class AccordJournalValueSerializers
@Override
public void deserialize(JournalKey journalKey, DurableBeforeAccumulator into, DataInputPlus in, Version userVersion) throws IOException
{
- // TODO: maybe using local serializer is not the best call here, but how do we distinguish
- // between messaging and disk versioning?
into.update(CommandStoreSerializers.durableBefore.deserialize(in));
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index e87250fa52..e597d7d415 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -129,7 +129,6 @@ public class AccordKeyspace
public static final Set<String> TABLE_NAMES = ImmutableSet.of(COMMANDS_FOR_KEY, JOURNAL);
- // TODO (desired): implement a custom type so we can get correct sort order
public static final TupleType TIMESTAMP_TYPE = new TupleType(Lists.newArrayList(LongType.instance, LongType.instance, Int32Type.instance));
private static final ClusteringIndexFilter FULL_PARTITION = new ClusteringIndexNamesFilter(BTreeSet.of(new ClusteringComparator(), Clustering.EMPTY), false);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
index 5ace8976c8..bc7c7cf841 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
@@ -225,7 +225,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
commandsForRanges.visit(keysOrRanges, startedBefore, testKind, visitor, p1, p2);
}
- // TODO (expected): instead of accepting a slice, accept the min/max epoch and let implementation handle it
@Override
public boolean visit(Unseekables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, Timestamp testStartedAtTimestamp, ComputeIsDep computeIsDep, AllCommandVisitor visit)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index 48c92e6296..f85aee62af 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -467,7 +467,6 @@ public class AccordService implements IAccordService, Shutdownable
if (response.current >= from)
return response;
- metadata = ClusterMetadata.current();
}
catch (Throwable e)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java
index 61d86db9bf..9930d2093e 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTask.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java
@@ -194,11 +194,11 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
private final String loggingId;
private static final AtomicLong nextLoggingId = new AtomicLong(Clock.Global.currentTimeMillis());
- // TODO (expected): merge all of these maps into one
+ // TODO (desired): merge all of these maps into one
@Nullable Object2ObjectHashMap<TxnId, AccordSafeCommand> commands;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeCommandsForKey> commandsForKey;
@Nullable Object2ObjectHashMap<Object, AccordSafeState<?, ?>> loading;
- // TODO (expected): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
+ // TODO (desired): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
@Nullable ArrayDeque<AccordCacheEntry<?, ?>> waitingToLoad;
@Nullable RangeTxnScanner rangeScanner;
boolean hasRanges;
@@ -662,7 +662,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
safeStore = commandStore.begin(this, commandsForRanges);
R result = apply(safeStore);
- // TODO (required): currently, we are not very efficient about ensuring that we persist the absolute minimum amount of state. Improve that.
List<Journal.CommandUpdate> changes = null;
if (commands != null)
{
diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
index ffd1754d2c..470822906f 100644
--- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
+++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
@@ -46,7 +46,7 @@ import org.apache.cassandra.service.accord.api.TokenKey;
import static accord.local.CommandSummaries.SummaryStatus.NOT_DIRECTLY_WITNESSED;
-// TODO (required): move to accord-core, merge with existing logic there
+// TODO (expected): move to accord-core, merge with existing logic there
public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements CommandSummaries.ByTxnIdSnapshot
{
public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
index 91acc361d1..ca466105eb 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
@@ -31,6 +31,7 @@ import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.coordinate.ExecutePath;
import accord.local.Node;
import accord.messages.Apply;
+import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Route;
@@ -86,23 +87,23 @@ public class AccordInteropAdapter extends TxnAdapter
}
@Override
- public void execute(Node node, Topologies any, FullRoute<?> route, ExecutePath path, ExecuteFlags executeFlags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
+ public void execute(Node node, Topologies any, FullRoute<?> route, Ballot ballot, ExecutePath path, ExecuteFlags executeFlags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
{
- if (!doInteropExecute(node, route, txnId, txn, executeAt, stableDeps, callback))
- super.execute(node, any, route, path, executeFlags, txnId, txn, executeAt, stableDeps, sendDeps, callback);
+ if (!doInteropExecute(node, route, ballot, txnId, txn, executeAt, stableDeps, callback))
+ super.execute(node, any, route, ballot, path, executeFlags, txnId, txn, executeAt, stableDeps, sendDeps, callback);
}
@Override
- public void persist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, FullRoute<?> route, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
+ public void persist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
{
- if (applyKind == Minimal && doInteropPersist(node, any, require, sendTo, selectSendTo, txnId, txn, executeAt, deps, writes, result, route, callback))
+ if (applyKind == Minimal && doInteropPersist(node, any, require, sendTo, selectSendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, callback))
return;
- super.persist(node, any, require, sendTo, selectSendTo, route, txnId, txn, executeAt, deps, writes, result, callback);
+ super.persist(node, any, require, sendTo, selectSendTo, route, ballot, txnId, txn, executeAt, deps, writes, result, callback);
}
- private boolean doInteropExecute(Node node, FullRoute<?> route, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
+ private boolean doInteropExecute(Node node, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
{
// Unrecoverable repair always needs to be run by AccordInteropExecution
AccordUpdate.Kind updateKind = AccordUpdate.kind(txn.update());
@@ -110,12 +111,12 @@ public class AccordInteropAdapter extends TxnAdapter
if (updateKind != AccordUpdate.Kind.UNRECOVERABLE_REPAIR && (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ONE || txn.read().keys().isEmpty()))
return false;
- new AccordInteropExecution(node, txnId, txn, updateKind, route, executeAt, deps, callback, executor, consistencyLevel, endpointMapper)
+ new AccordInteropExecution(node, txnId, txn, updateKind, route, ballot, executeAt, deps, callback, executor, consistencyLevel, endpointMapper)
.start();
return true;
}
- private boolean doInteropPersist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, BiConsumer<? super Result, Throwable> callback)
+ private boolean doInteropPersist(Node node, Topologies any, Route<?> require, Route<?> sendTo, SelectNodeOwnership selectSendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, BiConsumer<? super Result, Throwable> callback)
{
Update update = txn.update();
ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null;
@@ -123,7 +124,7 @@ public class AccordInteropAdapter extends TxnAdapter
return false;
Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt);
- new AccordInteropPersist(node, all, txnId, require, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, callback)
+ new AccordInteropPersist(node, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, callback)
.start(Minimal, any, writes, result);
return true;
}
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
index b9af95db34..3a5671c970 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
@@ -29,6 +29,7 @@ import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
import accord.messages.Apply;
import accord.messages.MessageType;
+import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
@@ -65,23 +66,23 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
public static final Apply.Factory FACTORY = new Apply.Factory()
{
@Override
- public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
+ public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Ballot ballot, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
{
checkArgument(kind != Kind.Maximal, "Shouldn't need to send a maximal commit with interop support");
ConsistencyLevel commitCL = txn.update() instanceof AccordUpdate ? ((AccordUpdate) txn.update()).cassandraCommitCL() : null;
// Any asynchronous apply option should use the regular Apply that doesn't wait for writes to complete
if (commitCL == null || commitCL == ConsistencyLevel.ANY)
- return Apply.FACTORY.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
- return new AccordInteropApply(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
+ return Apply.FACTORY.create(kind, to, participates, txnId, ballot, route, txn, executeAt, deps, writes, result, fullRoute);
+ return new AccordInteropApply(kind, to, participates, txnId, ballot, route, txn, executeAt, deps, writes, result, fullRoute);
}
};
public static final IVersionedSerializer<AccordInteropApply> serializer = new ApplySerializer<AccordInteropApply>()
{
@Override
- protected AccordInteropApply deserializeApply(TxnId txnId, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
+ protected AccordInteropApply deserializeApply(TxnId txnId, Ballot ballot, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
- return new AccordInteropApply(kind, txnId, scope, minEpoch, waitForEpoch, maxEpoch, executeAt, deps, txn, fullRoute, writes, result);
+ return new AccordInteropApply(kind, txnId, ballot, scope, minEpoch, waitForEpoch, maxEpoch, executeAt, deps, txn, fullRoute, writes, result);
}
};
@@ -89,14 +90,14 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
transient Int2ObjectHashMap<LocalListeners.Registered> listeners;
boolean failed;
- private AccordInteropApply(Kind kind, TxnId txnId, Route<?> route, long minEpoch, long waitForEpoch, long maxEpoch, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
+ private AccordInteropApply(Kind kind, TxnId txnId, Ballot ballot, Route<?> route, long minEpoch, long waitForEpoch, long maxEpoch, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
- super(kind, txnId, route, minEpoch, waitForEpoch, maxEpoch, executeAt, deps, txn, fullRoute, writes, result);
+ super(kind, txnId, ballot, route, minEpoch, waitForEpoch, maxEpoch, executeAt, deps, txn, fullRoute, writes, result);
}
- private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
+ private AccordInteropApply(Kind kind, Id to, Topologies participates, TxnId txnId, Ballot ballot, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
{
- super(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result, fullRoute);
+ super(kind, to, participates, txnId, ballot, route, txn, executeAt, deps, writes, result, fullRoute);
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
index 2d82cf1749..ffd42f0c48 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
@@ -144,6 +144,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
private final TxnId txnId;
private final Txn txn;
private final FullRoute<?> route;
+ private final Ballot ballot;
private final Timestamp executeAt;
private final Deps deps;
private final BiConsumer<? super Result, Throwable> callback;
@@ -161,7 +162,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
private final Set<InetAddressAndPort> contacted;
private final AccordUpdate.Kind updateKind;
- public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute<?> route, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback,
+ public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute<?> route, Ballot ballot, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback,
AgentExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper)
{
requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR);
@@ -169,6 +170,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
this.txnId = txnId;
this.txn = txn;
this.route = route;
+ this.ballot = ballot;
this.executeAt = executeAt;
this.deps = deps;
this.callback = callback;
@@ -402,7 +404,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
CommandStore cs = node.commandStores().select(route.homeKey());
result.beginAsResult().withExecutor(cs).begin((data, failure) -> {
if (failure == null)
- ((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, txnId, txn, executeAt, deps, txnId.is(Write) ? txn.execute(txnId, executeAt, data) : null, txn.result(txnId, executeAt, data), callback);
+ ((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, ballot, txnId, txn, executeAt, deps, txnId.is(Write) ? txn.execute(txnId, executeAt, data) : null, txn.result(txnId, executeAt, data), callback);
else
callback.accept(null, failure);
});
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
index e61d3934b5..967438ea56 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
@@ -28,6 +28,7 @@ import accord.coordinate.tracking.RequestStatus;
import accord.coordinate.tracking.ResponseTracker;
import accord.local.Node;
import accord.messages.Apply;
+import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Route;
@@ -108,9 +109,9 @@ public class AccordInteropPersist extends Persist
private final ConsistencyLevel consistencyLevel;
private CallbackHolder callback;
- public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, BiConsumer<? super Result, Throwable> clientCallback)
+ public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Ballot ballot, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, BiConsumer<? super Result, Throwable> clientCallback)
{
- super(node, topologies, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute, AccordInteropApply.FACTORY);
+ super(node, topologies, txnId, ballot, sendTo, txn, executeAt, deps, writes, result, fullRoute, AccordInteropApply.FACTORY);
Invariants.requireArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE);
this.consistencyLevel = consistencyLevel;
registerClientCallback(result, clientCallback);
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
index cf3f449354..181634cc50 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import accord.api.Result;
import accord.messages.Apply;
+import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
@@ -65,6 +66,7 @@ public class ApplySerializers
@Override
public void serializeBody(A apply, DataOutputPlus out, Version version) throws IOException
{
+ CommandSerializers.ballot.serialize(apply.ballot, out);
out.writeVInt(apply.minEpoch - apply.waitForEpoch);
out.writeUnsignedVInt(apply.maxEpoch - apply.minEpoch);
kind.serialize(apply.kind, out);
@@ -76,15 +78,16 @@ public class ApplySerializers
CommandSerializers.writes.serialize(apply.writes, out, version);
}
- protected abstract A deserializeApply(TxnId txnId, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind,
+ protected abstract A deserializeApply(TxnId txnId, Ballot ballot, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result);
@Override
public A deserializeBody(DataInputPlus in, Version version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
+ Ballot ballot = CommandSerializers.ballot.deserialize(in);
long minEpoch = waitForEpoch + in.readVInt();
long maxEpoch = minEpoch + in.readUnsignedVInt();
- return deserializeApply(txnId, scope, minEpoch, waitForEpoch, maxEpoch,
+ return deserializeApply(txnId, ballot, scope, minEpoch, waitForEpoch, maxEpoch,
kind.deserialize(in),
ExecuteAtSerializer.deserialize(txnId, in),
DepsSerializers.partialDeps.deserialize(in),
@@ -97,7 +100,8 @@ public class ApplySerializers
@Override
public long serializedBodySize(A apply, Version version)
{
- return TypeSizes.sizeofVInt(apply.minEpoch - apply.waitForEpoch)
+ return CommandSerializers.ballot.serializedSize(apply.ballot)
+ + TypeSizes.sizeofVInt(apply.minEpoch - apply.waitForEpoch)
+ TypeSizes.sizeofUnsignedVInt(apply.maxEpoch - apply.minEpoch)
+ kind.serializedSize(apply.kind)
+ ExecuteAtSerializer.serializedSize(apply.txnId, apply.executeAt)
@@ -111,10 +115,10 @@ public class ApplySerializers
public static final IVersionedSerializer<Apply> request = new ApplySerializer<>()
{
@Override
- protected Apply deserializeApply(TxnId txnId, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind,
+ protected Apply deserializeApply(TxnId txnId, Ballot ballot, Route<?> scope, long minEpoch, long waitForEpoch, long maxEpoch, Apply.Kind kind,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result)
{
- return Apply.SerializationSupport.create(txnId, scope, minEpoch, waitForEpoch, maxEpoch, kind, executeAt, deps, txn, fullRoute, writes, result);
+ return Apply.SerializationSupport.create(txnId, ballot, scope, minEpoch, waitForEpoch, maxEpoch, kind, executeAt, deps, txn, fullRoute, writes, result);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
index 9933f53f9b..6981761612 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
@@ -130,6 +130,7 @@ public class CheckStatusSerializers
KeySerializers.participants.serialize(check.query, out);
out.writeUnsignedVInt(check.sourceEpoch);
out.writeByte(check.includeInfo.ordinal());
+ CommandSerializers.ballot.serialize(check.bumpBallot, out);
}
@Override
@@ -139,7 +140,8 @@ public class CheckStatusSerializers
Participants<?> query = KeySerializers.participants.deserialize(in);
long sourceEpoch = in.readUnsignedVInt();
CheckStatus.IncludeInfo info = infos[in.readByte()];
- return new CheckStatus(txnId, query, sourceEpoch, info);
+ Ballot ballot = CommandSerializers.ballot.deserialize(in);
+ return new CheckStatus(txnId, query, sourceEpoch, info, ballot);
}
@Override
@@ -148,7 +150,8 @@ public class CheckStatusSerializers
return CommandSerializers.txnId.serializedSize(check.txnId)
+ KeySerializers.participants.serializedSize(check.query)
+ TypeSizes.sizeofUnsignedVInt(check.sourceEpoch)
- + TypeSizes.BYTE_SIZE;
+ + TypeSizes.BYTE_SIZE
+ + CommandSerializers.ballot.serializedSize(check.bumpBallot);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java
index 3aedf090ea..82e702b5c2 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java
@@ -223,7 +223,6 @@ public class ReadDataSerializers
public static final class ReplySerializer<D extends Data> implements IVersionedSerializer<ReadReply>
{
- // TODO (expected): use something other than ordinal
final CommitOrReadNack[] nacks = CommitOrReadNack.values();
private final VersionedSerializer<D, Version> dataSerializer;
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java
index 0e3413905f..5d2d5efa46 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java
@@ -26,7 +26,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
public class ResultSerializers
{
- // TODO (expected): this is meant to encode e.g. whether the transaction's condition met or not for clients to later query
+ // TODO (desired): this is meant to encode e.g. whether the transaction's condition met or not for clients to later query
public static final Result APPLIED = new Result()
{
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
index 800a491b6d..fe2cbe2613 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
@@ -52,7 +52,7 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
{
TxnId txnId = CommandSerializers.txnId.deserialize(in);
Route<?> scope = KeySerializers.route.deserialize(in);
- // TODO: there should be a base epoch
+ // TODO (desired): there should be a base epoch
long waitForEpoch = in.readUnsignedVInt();
return deserializeBody(in, version, txnId, scope, waitForEpoch);
}
diff --git a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java
index 1bf46889dd..efb222b655 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java
@@ -78,7 +78,6 @@ public class AccordUpdateParameters
// For the time being, guardrails are disabled for Accord queries.
ClientState disabledGuardrails = null;
- // TODO : How should Accord work with TTL?
int ttl = metadata.params.defaultTimeToLive;
return new RowUpdateParameters(metadata,
disabledGuardrails,
@@ -103,7 +102,6 @@ public class AccordUpdateParameters
checkState(data.entrySet().size() == 1, "CAS read should only have one entry");
return ImmutableMap.of(dk, value);
case AUTO_READ:
- // TODO (review): Is this the right DK being passed into that matches what we used to store in TxnDataName
if (TxnData.txnDataNameIndex(name) == index)
return ImmutableMap.of(dk, value);
default:
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
index 2f84c849e6..ea27a398c2 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
@@ -235,9 +235,6 @@ public class TxnNamedRead extends AbstractSerialized<ReadCommand, TableMetadatas
if (command == null)
return AsyncResults.success(TxnData.NOOP_DATA);
- // TODO (required, safety): before release, double check reasoning that this is safe
-// AccordCommandsForKey cfk = ((SafeAccordCommandStore)safeStore).commandsForKey(key);
-// int nowInSeconds = cfk.nowInSecondsFor(executeAt, isForWriteTxn);
// It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
// this simply looks like the transaction witnessed TTL'd data and the data then expired
// immediately after the transaction executed, and this simplifies things a great deal
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
index 8f3baf374e..1fdc0e54f9 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
@@ -174,7 +174,7 @@ public class TxnUpdate extends AccordUpdate
public Update slice(Ranges ranges)
{
Keys keys = this.keys.slice(ranges);
- // TODO: Slice the condition.
+ // TODO (desired): Slice the condition.
return new TxnUpdate(tables, keys, select(this.keys, keys, fragments), condition, cassandraCommitCL, preserveTimestamps);
}
@@ -182,7 +182,7 @@ public class TxnUpdate extends AccordUpdate
public Update intersecting(Participants<?> participants)
{
Keys keys = this.keys.intersecting(participants);
- // TODO: Slice the condition.
+ // TODO (desired): Slice the condition.
return new TxnUpdate(tables, keys, select(this.keys, keys, fragments), condition, cassandraCommitCL, preserveTimestamps);
}
@@ -201,9 +201,9 @@ public class TxnUpdate extends AccordUpdate
@Override
public Update merge(Update update)
{
- // TODO: special method for linear merging keyed and non-keyed lists simultaneously
TxnUpdate that = (TxnUpdate) update;
Keys mergedKeys = this.keys.with(that.keys);
+ // TODO (desired): special method for linear merging keyed and non-keyed lists simultaneously
ByteBuffer[] mergedFragments = merge(this.keys, that.keys, this.fragments, that.fragments, mergedKeys.size());
return new TxnUpdate(tables, mergedKeys, mergedFragments, condition, cassandraCommitCL, preserveTimestamps);
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/HintsMaxSizeTest.java b/test/distributed/org/apache/cassandra/distributed/test/HintsMaxSizeTest.java
index b1cf19e703..7e25f4311d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/HintsMaxSizeTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/HintsMaxSizeTest.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.distributed.test;
import java.util.UUID;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.auth.CassandraRoleManager;
@@ -44,7 +43,6 @@ import static org.awaitility.Awaitility.await;
@SuppressWarnings("Convert2MethodRef")
public class HintsMaxSizeTest extends TestBaseImpl
{
- @Ignore
@Test
public void testMaxHintedHandoffSize() throws Exception
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java b/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java
index b3de4edbd6..ea3c980586 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java
@@ -26,7 +26,6 @@ import com.google.common.collect.ImmutableMap;
import org.apache.commons.io.FileUtils;
import org.junit.AfterClass;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -106,7 +105,6 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
assertRows(CLUSTER.get(1).executeInternal("SELECT count(*) FROM ssl_upload_tables.test"), row(42L));
}
- @Ignore
@Test
public void bulkLoaderSuccessfullyStreamsOverSslWithDeprecatedSslStoragePort() throws Throwable
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java b/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
index d42c379683..78c67c9e15 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
@@ -90,7 +90,9 @@ public class FetchLogFromPeers2Test extends TestBaseImpl
cluster.coordinator(coordinator).execute(withKeyspace(query), ConsistencyLevel.QUORUM);
fail("should fail");
}
- catch (Exception ignored) {}
+ catch (Exception ignored)
+ {
+ }
boolean metricBumped = false;
for (int i = 1; i <= cluster.size(); i++)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tcm/SplitBrainTest.java b/test/distributed/org/apache/cassandra/distributed/test/tcm/SplitBrainTest.java
index 9e35a71d83..1382f8b063 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tcm/SplitBrainTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tcm/SplitBrainTest.java
@@ -30,7 +30,6 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
@@ -57,7 +56,6 @@ import static org.junit.Assert.assertTrue;
public class SplitBrainTest extends TestBaseImpl
{
- @Ignore
@Test
public void testSplitBrainStartup() throws IOException, TimeoutException
{
diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
index 14d1587c5b..0c0f43dc2d 100644
--- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
+++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
@@ -40,11 +40,17 @@ import accord.impl.TopologyFactory;
import accord.impl.basic.Cluster;
import accord.impl.basic.RandomDelayQueue;
import accord.local.CommandStores;
+import accord.local.DurableBefore;
import accord.local.Node;
+import accord.local.RedundantBefore;
import accord.primitives.EpochSupplier;
import accord.utils.DefaultRandom;
import accord.utils.Invariants;
+import accord.utils.PersistentField;
import accord.utils.RandomSource;
+import accord.utils.async.AsyncChains;
+import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
@@ -308,6 +314,13 @@ public class AccordJournalBurnTest extends BurnTestBase
this.closeCurrentSegmentForTestingIfNonEmpty();
super.replay(commandStores);
}
+
+ @Override
+ public PersistentField.Persister<DurableBefore, DurableBefore> durableBeforePersister()
+ {
+ // TODO (required): we should be persisting in the journal, but this currently causes the burn test to take far too long
+ return DurableBefore.NOOP_PERSISTER;
+ }
};
return journal;
@@ -330,9 +343,15 @@ public class AccordJournalBurnTest extends BurnTestBase
{
IAccordService.AccordCompactionInfos compactionInfos = new IAccordService.AccordCompactionInfos(node.durableBefore(), node.topology().minEpoch());
node.commandStores().forEachCommandStore(commandStore -> {
+ RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
+ if (redundantBefore == null)
+ redundantBefore = RedundantBefore.EMPTY;
+ CommandStores.RangesForEpoch rangesForEpoch = commandStore.unsafeGetRangesForEpoch();
+ if (rangesForEpoch == null)
+ rangesForEpoch = CommandStores.RangesForEpoch.EMPTY;
compactionInfos.put(commandStore.id(), new IAccordService.AccordCompactionInfo(commandStore.id(),
- commandStore.unsafeGetRedundantBefore(),
- commandStore.unsafeGetRangesForEpoch(),
+ redundantBefore,
+ rangesForEpoch,
tableId));
});
return compactionInfos;
diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java
index a9f53a340e..702e2b6871 100644
--- a/test/unit/org/apache/cassandra/ServerTestUtils.java
+++ b/test/unit/org/apache/cassandra/ServerTestUtils.java
@@ -238,7 +238,7 @@ public final class ServerTestUtils
{
if (directory.exists())
{
- Arrays.stream(directory.tryList()).forEach(File::deleteRecursive);
+ Arrays.stream(directory.tryList()).forEach(File::tryDeleteRecursive);
}
}
diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
index 08b9df49f1..f23e16788e 100644
--- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
@@ -33,7 +33,6 @@ import org.apache.cassandra.io.util.FileReader;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -110,7 +109,6 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
testWithNonblockingMode(this::testSegmentFlaggingOnCreation0);
}
- @Ignore
@Test
public void testNonblockingShouldMaintainSteadyDiskUsage() throws Throwable
{
diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
index becc9390dd..e3e5662112 100644
--- a/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
@@ -27,7 +27,6 @@ import java.util.Random;
import com.google.common.collect.ImmutableMap;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -58,7 +57,6 @@ public class CommitlogShutdownTest
private final static byte[] entropy = new byte[1024 * 256];
- @Ignore
@Test
@BMRule(name = "Make removing commitlog segments slow",
targetClass = "CommitLogSegment",
diff --git a/test/unit/org/apache/cassandra/hints/HintServiceBytemanTest.java b/test/unit/org/apache/cassandra/hints/HintServiceBytemanTest.java
index e44f8dd973..1beaa123a8 100644
--- a/test/unit/org/apache/cassandra/hints/HintServiceBytemanTest.java
+++ b/test/unit/org/apache/cassandra/hints/HintServiceBytemanTest.java
@@ -26,7 +26,6 @@ import java.util.concurrent.ExecutionException;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -92,7 +91,6 @@ public class HintServiceBytemanTest
HintsService.instance.startDispatch();
}
- @Ignore
@Test
@BMRule(name = "Delay delivering hints",
targetClass = "DispatchHintsTask",
diff --git a/test/unit/org/apache/cassandra/io/util/SafeMemoryWriterTest.java b/test/unit/org/apache/cassandra/io/util/SafeMemoryWriterTest.java
index fd5075a9cd..8b37c2def2 100644
--- a/test/unit/org/apache/cassandra/io/util/SafeMemoryWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/util/SafeMemoryWriterTest.java
@@ -22,7 +22,6 @@ import java.util.Arrays;
import java.util.Random;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -60,7 +59,6 @@ public class SafeMemoryWriterTest
testSafeMemoryWriter(CHUNK * 5, CHUNK, 65536);
}
- @Ignore
@Test
public void testOver2GBuffer() throws IOException
{
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java
index 7079de790e..5a8ac1505b 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java
@@ -69,12 +69,12 @@ public class CheckStatusSerializersTest
switch (domain)
{
case Key:
- // TODO (coverage): don't hard code murmur
+ // TODO (desired): don't hard code murmur
Gen<TokenKey> keyGen = AccordGenerators.routingKeyGen(fromQT(CassandraGenerators.TABLE_ID_GEN), Gens.constant(AccordGenerators.RoutingKeyKind.TOKEN), fromQT(CassandraGenerators.murmurToken()), Murmur3Partitioner.instance);
TokenKey homeKey = keyGen.next(rs);
List<TokenKey> forOrdering = Gens.lists(keyGen).unique().ofSizeBetween(1, 10).next(rs);
forOrdering.sort(Comparator.naturalOrder());
- // TODO (coverage): don't hard code keys type
+ // TODO (desired): don't hard code keys type
keysOrRanges = new FullKeyRoute(homeKey, forOrdering.toArray(RoutingKey[]::new));
break;
case Range:
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
index 90241b7475..dcea11348d 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
@@ -479,7 +479,6 @@ public class CommandsForKeySerializerTest
}
}
- // TODO (expected): we currently don't explore TruncatedApply statuses because we don't transition through all phases and therefore don't adopt the Applied status
Choices<SaveStatus> saveStatusChoices = Choices.uniform(EnumSet.complementOf(EnumSet.of(SaveStatus.TruncatedApply, SaveStatus.TruncatedUnapplied, SaveStatus.TruncatedApplyWithOutcome)).toArray(SaveStatus[]::new));
Supplier<SaveStatus> saveStatusSupplier = () -> {
SaveStatus result = saveStatusChoices.choose(source);
diff --git a/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java b/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java
index eae6ea2f62..6021355442 100644
--- a/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java
+++ b/test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java
@@ -32,7 +32,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.NotImplementedException;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -78,7 +77,6 @@ public class DiscoverySimulationTest
log.readyUnchecked();
}
- @Ignore
@Test
public void discoveryTest() throws Throwable
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org