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/11/27 13:12:38 UTC
(cassandra) branch trunk updated (6ab1ccf2fc -> 4defbb9b1c)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
omit 6ab1ccf2fc Make TopologyException a checked exception to ensure they are handled carefully, as they may occur at surprising times Also Fix: - Restore MaxDecidedRX on replay - When catchup_on_start_exit_on_failure == false, should startup on any kind of failure, not only timeout - lazy vtable LIMIT clause regression - DurabilityService.onEpochRetired - Command.validate when uniqueHlc differs - Avoid unsafe publication of AccordExecutor to scheduledFastTasks - AccordCache hi [...]
new 4defbb9b1c Make TopologyException a checked exception to ensure they are handled carefully, as they may occur at surprising times Also Fix: - Restore MaxDecidedRX on replay - When catchup_on_start_exit_on_failure == false, should startup on any kind of failure, not only timeout - lazy vtable LIMIT clause regression - DurabilityService.onEpochRetired - Command.validate when uniqueHlc differs - Avoid unsafe publication of AccordExecutor to scheduledFastTasks - AccordCache hi [...]
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (6ab1ccf2fc)
\
N -- N -- N refs/heads/trunk (4defbb9b1c)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitmodules | 4 ++--
modules/accord | 2 +-
2 files changed, 3 insertions(+), 3 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: Make TopologyException a checked exception to ensure they are handled carefully, as they may occur at surprising times Also Fix: - Restore MaxDecidedRX on replay - When catchup_on_start_exit_on_failure == false, should startup on any kind of failure, not only timeout - lazy vtable LIMIT clause regression - DurabilityService.onEpochRetired - Command.validate when uniqueHlc differs - Avoid unsafe publication of AccordExecutor to scheduledFastTasks - AccordCache hitRate metric names - [...]
Posted by be...@apache.org.
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
commit 4defbb9b1c9cc41e71f3cef0968b108a18cf88c2
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Nov 13 12:07:51 2025 +0000
Make TopologyException a checked exception to ensure they are handled carefully, as they may occur at surprising times
Also Fix:
- Restore MaxDecidedRX on replay
- When catchup_on_start_exit_on_failure == false, should startup on any kind of failure, not only timeout
- lazy vtable LIMIT clause regression
- DurabilityService.onEpochRetired
- Command.validate when uniqueHlc differs
- Avoid unsafe publication of AccordExecutor to scheduledFastTasks
- AccordCache hitRate metric names
- use long for return type of DurationSpec.toNanoseconds
- Repair without all replicas should not request all Accord replicas participate
- ExecuteAtSerializer
- SyncPoints should be coordinated in an epoch that contains the ranges
Also Improve:
- Split Accord startup into local+distributed, ensure we
- Add logging to FetchDurableBefore on startup
- Add randomised testing of AbstractLazyVirtualTable
- Add validation of lazy virtual table key ordering
- Don't send requests to faulty replicas
- shrinkOrEvict large objects without holding lock
- Accord dtest shutdown
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21042
---
modules/accord | 2 +-
.../config/CassandraRelevantProperties.java | 1 +
.../org/apache/cassandra/config/DurationSpec.java | 8 +-
.../db/compaction/CompactionIterator.java | 11 +-
.../db/virtual/AbstractLazyVirtualTable.java | 93 ++++--
.../cassandra/db/virtual/AccordDebugKeyspace.java | 4 +-
.../apache/cassandra/journal/ActiveSegment.java | 2 +-
.../org/apache/cassandra/journal/Compactor.java | 6 +
src/java/org/apache/cassandra/journal/Flusher.java | 2 +
src/java/org/apache/cassandra/journal/Journal.java | 4 +-
.../cassandra/metrics/AccordCacheMetrics.java | 6 +-
.../cassandra/metrics/AccordMetricUtils.java | 4 +-
.../cassandra/metrics/AccordSystemMetrics.java | 6 +-
.../org/apache/cassandra/net/MessageDelivery.java | 2 +-
src/java/org/apache/cassandra/net/Verb.java | 10 +-
.../cassandra/repair/AbstractRepairTask.java | 7 +-
.../cassandra/repair/IncrementalRepairTask.java | 6 +-
.../apache/cassandra/repair/NormalRepairTask.java | 13 +-
.../apache/cassandra/repair/PreviewRepairTask.java | 10 +-
.../apache/cassandra/repair/RepairCoordinator.java | 42 ++-
.../org/apache/cassandra/repair/RepairJob.java | 24 +-
.../org/apache/cassandra/repair/RepairSession.java | 9 +-
.../org/apache/cassandra/repair/SharedContext.java | 2 +-
.../repair/autorepair/AutoRepairState.java | 6 +-
.../cassandra/repair/messages/RepairOption.java | 14 +-
.../cassandra/repair/state/CoordinatorState.java | 2 +-
.../cassandra/service/ActiveRepairService.java | 36 ++-
.../apache/cassandra/service/CassandraDaemon.java | 7 +-
.../apache/cassandra/service/StorageService.java | 18 +-
.../cassandra/service/accord/AccordCache.java | 90 +++++-
.../cassandra/service/accord/AccordCacheEntry.java | 39 ++-
.../service/accord/AccordCommandStore.java | 5 +
.../service/accord/AccordCommandStores.java | 12 +
.../cassandra/service/accord/AccordDataStore.java | 11 +-
.../cassandra/service/accord/AccordExecutor.java | 22 +-
.../accord/AccordExecutorAbstractLockLoop.java | 8 +-
.../service/accord/AccordFetchCoordinator.java | 11 +-
.../cassandra/service/accord/AccordJournal.java | 3 +-
.../accord/AccordJournalValueSerializers.java | 1 -
.../service/accord/AccordMessageSink.java | 19 +-
.../service/accord/AccordResponseVerbHandler.java | 6 -
.../cassandra/service/accord/AccordResult.java | 42 +--
.../cassandra/service/accord/AccordService.java | 100 ++++--
.../cassandra/service/accord/AccordTask.java | 10 +-
.../cassandra/service/accord/AccordTracing.java | 3 +
.../service/accord/AccordVerbHandler.java | 6 -
.../cassandra/service/accord/IAccordService.java | 23 +-
.../service/accord/ImmediateAsyncExecutor.java | 2 +-
.../service/accord/WatermarkCollector.java | 1 -
.../cassandra/service/accord/api/AccordAgent.java | 21 +-
.../accord/interop/AccordInteropAdapter.java | 30 +-
.../service/accord/interop/AccordInteropApply.java | 2 +-
.../accord/interop/AccordInteropExecution.java | 3 +-
.../service/accord/repair/AccordRepair.java | 7 +-
.../serializers/AbstractSortedCollector.java | 30 +-
.../accord/serializers/CommandSerializers.java | 2 +-
.../cassandra/service/accord/txn/TxnUpdate.java | 5 +-
.../cassandra/service/accord/txn/TxnWrite.java | 3 +-
.../migration/ConsensusTableMigration.java | 2 +-
.../apache/cassandra/tools/nodetool/Repair.java | 4 +
.../org/apache/cassandra/utils/ExecutorUtils.java | 41 ++-
.../distributed/impl/AbstractCluster.java | 27 +-
.../cassandra/distributed/impl/Instance.java | 18 +-
.../distributed/test/DecommissionTest.java | 7 +
.../distributed/test/accord/AccordCQLTestBase.java | 5 +-
.../AccordCommandStoreTryExecuteListeningTest.java | 5 +-
.../distributed/test/accord/AccordMetricsTest.java | 1 -
.../test/accord/AccordSimpleFastPathTest.java | 2 +-
.../accord/journal/AccordJournalReplayTest.java | 5 +-
test/resources/nodetool/help/repair | 8 +-
.../simulator/cluster/OnInstanceRepair.java | 2 +-
.../compaction/CompactionAccordIteratorsTest.java | 8 +-
.../db/virtual/AccordDebugKeyspaceTest.java | 36 ++-
.../cassandra/db/virtual/LazyVirtualTableTest.java | 351 +++++++++++++++++++++
.../db/virtual/LocalRepairTablesTest.java | 2 +-
.../cassandra/index/accord/RouteIndexTest.java | 3 +-
.../cassandra/repair/NeighborsAndRangesTest.java | 8 +-
.../org/apache/cassandra/repair/RepairJobTest.java | 4 +-
.../apache/cassandra/repair/RepairSessionTest.java | 4 +-
.../autorepair/AutoRepairParameterizedTest.java | 6 -
.../cassandra/service/accord/AccordCacheTest.java | 16 +-
.../service/accord/AccordCommandTest.java | 4 +-
.../service/accord/AccordMessageSinkTest.java | 3 +-
.../cassandra/service/accord/AccordTaskTest.java | 106 ++++---
.../cassandra/service/accord/EpochSyncTest.java | 4 +-
.../accord/SimulatedAccordCommandStore.java | 6 +-
.../serializers/CommandsForKeySerializerTest.java | 4 +-
.../accord/serializers/KeySerializersTest.java | 2 +-
.../service/accord/txn/TxnUpdateTest.java | 6 +-
.../paxos/uncommitted/PaxosStateTrackerTest.java | 2 +-
90 files changed, 1145 insertions(+), 431 deletions(-)
diff --git a/modules/accord b/modules/accord
index 938ba19ada..973335bfe7 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 938ba19adaf70bf9901fb2dc177dce313d7ee15c
+Subproject commit 973335bfe7e930c6646016a4a48cff084f49b660
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 2bcf423c7e..fc1c1c1462 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -228,6 +228,7 @@ public enum CassandraRelevantProperties
DTEST_ACCORD_ENABLED("jvm_dtest.accord.enabled", "true"),
DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED("jvm_dtest.accord.journal_sanity_check_enabled", "false"),
DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
+ DTEST_IGNORE_SHUTDOWN_THREADCOUNT("jvm_dtests.ignore_shutdown_threadcount"),
/** This property indicates if the code is running under the in-jvm dtest framework */
DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
/** In_JVM dtest property indicating that the test should use "latest" configuration */
diff --git a/src/java/org/apache/cassandra/config/DurationSpec.java b/src/java/org/apache/cassandra/config/DurationSpec.java
index 01a7d70b5f..8c90542eb8 100644
--- a/src/java/org/apache/cassandra/config/DurationSpec.java
+++ b/src/java/org/apache/cassandra/config/DurationSpec.java
@@ -587,13 +587,11 @@ public abstract class DurationSpec
}
/**
- * Returns this duration in the number of nanoseconds as an {@code int}
- *
- * @return this duration in number of nanoseconds or {@code Integer.MAX_VALUE} if the number of nanoseconds is too large.
+ * Returns this duration in the number of nanoseconds as a {@code long}
*/
- public int toNanoseconds()
+ public long toNanoseconds()
{
- return Ints.saturatedCast(unit().toNanos(quantity()));
+ return unit().toNanos(quantity());
}
/**
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 8facfe30da..be4ad3d921 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -181,7 +181,14 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
TopPartitionTracker.Collector topPartitionCollector)
{
this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector,
- AccordService.isSetup() ? AccordService.instance() : null);
+ accord(controller));
+ }
+
+ private static IAccordService accord(AbstractCompactionController controller)
+ {
+ IAccordService accord = AccordService.tryGetUnsafe();
+ Invariants.require(accord != null || (!isAccordJournal(controller.cfs) && !isAccordCommandsForKey(controller.cfs)));
+ return accord;
}
public CompactionIterator(OperationType type,
@@ -194,7 +201,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
IAccordService accord)
{
this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector,
- () -> accord.getCompactionInfo(),
+ () -> Invariants.nonNull(accord).getCompactionInfo(),
() -> Version.fromVersion(accord.journalConfiguration().userVersion()));
}
diff --git a/src/java/org/apache/cassandra/db/virtual/AbstractLazyVirtualTable.java b/src/java/org/apache/cassandra/db/virtual/AbstractLazyVirtualTable.java
index 6377824417..68b010da58 100644
--- a/src/java/org/apache/cassandra/db/virtual/AbstractLazyVirtualTable.java
+++ b/src/java/org/apache/cassandra/db/virtual/AbstractLazyVirtualTable.java
@@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;
+import java.util.NoSuchElementException;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
@@ -223,6 +224,8 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (!dataRange.contains(partitionKey))
return dropCks -> {};
+
+
return partitions.computeIfAbsent(partitionKey, SimplePartition::new);
}
@@ -232,36 +235,50 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
final Iterator<SimplePartition> partitions = this.partitions.values().iterator();
return new UnfilteredPartitionIterator()
{
+ private UnfilteredRowIterator next;
+
@Override public TableMetadata metadata() { return metadata; }
@Override public void close() {}
@Override
public boolean hasNext()
{
- return partitions.hasNext();
+ while (next == null && partitions.hasNext())
+ {
+ SimplePartition partition = partitions.next();
+ Iterator<Row> rows = partition.rows();
+
+ if (!rows.hasNext())
+ continue;
+
+ next = new UnfilteredRowIterator()
+ {
+ @Override public TableMetadata metadata() { return metadata; }
+ @Override public boolean isReverseOrder() { return dataRange.isReversed(); }
+ @Override public RegularAndStaticColumns columns() { return columnFilter.fetchedColumns(); }
+ @Override public DecoratedKey partitionKey() { return partition.key; }
+
+ @Override public Row staticRow() { return partition.staticRow(); }
+ @Override public boolean hasNext() { return rows.hasNext(); }
+ @Override public Unfiltered next() { return rows.next(); }
+
+ @Override public void close() {}
+ @Override public DeletionTime partitionLevelDeletion() { return DeletionTime.LIVE; }
+ @Override public EncodingStats stats() { return EncodingStats.NO_STATS; }
+ };
+ }
+ return next != null;
}
@Override
public UnfilteredRowIterator next()
{
- SimplePartition partition = partitions.next();
- Iterator<Row> rows = partition.rows();
+ if (!hasNext())
+ throw new NoSuchElementException();
- return new UnfilteredRowIterator()
- {
- @Override public TableMetadata metadata() { return metadata; }
- @Override public boolean isReverseOrder() { return dataRange.isReversed(); }
- @Override public RegularAndStaticColumns columns() { return columnFilter.fetchedColumns(); }
- @Override public DecoratedKey partitionKey() { return partition.key; }
-
- @Override public Row staticRow() { return partition.staticRow(); }
- @Override public boolean hasNext() { return rows.hasNext(); }
- @Override public Unfiltered next() { return rows.next(); }
-
- @Override public void close() {}
- @Override public DeletionTime partitionLevelDeletion() { return DeletionTime.LIVE; }
- @Override public EncodingStats stats() { return EncodingStats.NO_STATS; }
- };
+ UnfilteredRowIterator result = next;
+ next = null;
+ return result;
}
};
}
@@ -312,9 +329,23 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (!dataRange.contains(partitionKey) || !dataRange.clusteringIndexFilter(partitionKey).selects(clustering))
return drop -> {};
+ if (isSortedByPartitionKey)
+ checkCorrectlySorted(partitionKey);
+
return partitions.computeIfAbsent(partitionKey, SimplePartition::new).row(clustering);
}
+ private void checkCorrectlySorted(DecoratedKey newPartitionKey)
+ {
+ if (partitions.isEmpty())
+ return;
+
+ DecoratedKey prevKey = partitions.lastKey();
+ int c = metadata.partitionKeyType.compare(prevKey.getKey(), newPartitionKey.getKey());
+ if (dataRange.isReversed() ? c < 0 : c > 0)
+ throw new IllegalArgumentException(Arrays.toString(composePartitionKeys(prevKey, metadata)) + (dataRange.isReversed() ? " < " : " > ") + Arrays.toString(composePartitionKeys(newPartitionKey, metadata)));
+ }
+
private final class SimplePartition implements PartitionCollector, RowsCollector
{
private final DecoratedKey key;
@@ -323,6 +354,7 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
private int rowCount;
private SimpleRow staticRow;
private boolean dropRows;
+ private boolean isSortedAndFiltered = true;
private SimplePartition(DecoratedKey key)
{
@@ -346,6 +378,17 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
return row(decomposeClusterings(metadata, clusteringKeys));
}
+ private void checkCorrectlySorted(Clustering<?> newClustering)
+ {
+ if (rowCount == 0)
+ return;
+
+ Clustering<?> prevClustering = rows[rowCount - 1].clustering;
+ int c = metadata.comparator.compare(prevClustering, newClustering);
+ if (dataRange.isReversed() ? c <= 0 : c >= 0)
+ throw new IllegalArgumentException(Arrays.toString(composeClusterings(prevClustering, metadata)) + (dataRange.isReversed() ? " <= " : " >= ") + Arrays.toString(composeClusterings(newClustering, metadata)));
+ }
+
RowCollector row(Clustering<?> clustering)
{
if (nanoTime() > deadlineNanos)
@@ -354,6 +397,9 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (dropRows || !dataRange.clusteringIndexFilter(key).selects(clustering))
return drop -> {};
+ if (isSorted)
+ checkCorrectlySorted(clustering);
+
if (totalRowCount >= limits.count())
{
boolean filter;
@@ -370,9 +416,13 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (filter)
{
- // first filter within each partition
for (SimplePartition partition : partitions.values())
+ {
+ // first filter within each partition
+ partition.filterAndSort();
+ // and truncate if there are per-partition limits
partition.truncate(limits.perPartitionCount());
+ }
// then drop any partitions that completely fall outside our limit
Iterator<SimplePartition> iter = partitions.descendingMap().values().iterator();
@@ -418,12 +468,16 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (rowCount == rows.length)
rows = Arrays.copyOf(rows, Math.max(8, rowCount * 2));
rows[rowCount++] = result;
+ isSortedAndFiltered = false;
}
return result;
}
void filterAndSort()
{
+ if (isSortedAndFiltered)
+ return;
+
int newCount = 0;
for (int i = 0 ; i < rowCount; ++i)
{
@@ -441,6 +495,7 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
rowCount = newCount;
}
Arrays.sort(rows, 0, newCount, rowComparator());
+ isSortedAndFiltered = true;
}
int truncate(int newCount)
diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
index b32a166b8a..c0c4c82f29 100644
--- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
@@ -946,7 +946,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
collector.partition(commandStore.id()).collect(collect -> {
while (view.advance())
{
- // TODO (required): view should return an immutable per-row view so that we can call lazyAdd
+ // TODO (desired): view should return an immutable per-row view so that we can call lazyAdd
collect.add(view.txnId().toString())
.eagerCollect(columns -> {
columns.add("table_id", tableIdStr)
@@ -1516,7 +1516,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
AbstractJournalTable(TableMetadata metadata)
{
- super(metadata, FAIL, ASC);
+ super(metadata, FAIL, UNSORTED, ASC);
}
@Override
diff --git a/src/java/org/apache/cassandra/journal/ActiveSegment.java b/src/java/org/apache/cassandra/journal/ActiveSegment.java
index 5c93655552..75d3505c3f 100644
--- a/src/java/org/apache/cassandra/journal/ActiveSegment.java
+++ b/src/java/org/apache/cassandra/journal/ActiveSegment.java
@@ -466,7 +466,7 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
- // TODO (required): Find a better way to test unwritten allocations and/or corruption
+ // TODO (expected): Find a better way to test unwritten allocations and/or corruption
@VisibleForTesting
void consumeBufferUnsafe(Consumer<ByteBuffer> fn)
{
diff --git a/src/java/org/apache/cassandra/journal/Compactor.java b/src/java/org/apache/cassandra/journal/Compactor.java
index 330ecea0a9..7a1b19b088 100644
--- a/src/java/org/apache/cassandra/journal/Compactor.java
+++ b/src/java/org/apache/cassandra/journal/Compactor.java
@@ -24,6 +24,9 @@ import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.Shutdownable;
@@ -31,6 +34,8 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac
public final class Compactor<K, V> implements Runnable, Shutdownable
{
+ private static final Logger logger = LoggerFactory.getLogger(Compactor.class);
+
private final Journal<K, V> journal;
private final SegmentCompactor<K, V> segmentCompactor;
private final ScheduledExecutorPlus executor;
@@ -99,6 +104,7 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
@Override
public void shutdown()
{
+ logger.debug("Shutting down " + executor);
executor.shutdown();
}
diff --git a/src/java/org/apache/cassandra/journal/Flusher.java b/src/java/org/apache/cassandra/journal/Flusher.java
index 02f85df0cb..2ce4abc2ae 100644
--- a/src/java/org/apache/cassandra/journal/Flusher.java
+++ b/src/java/org/apache/cassandra/journal/Flusher.java
@@ -102,10 +102,12 @@ final class Flusher<K, V>
void shutdown() throws InterruptedException
{
+ logger.debug("Shutting down " + flushExecutor + " and awaiting termination");
flushExecutor.shutdown();
flushExecutor.awaitTermination(1, MINUTES);
if (fsyncExecutor != null)
{
+ logger.debug("Shutting down " + fsyncExecutor + " and awaiting termination");
fsyncExecutor.shutdownNow(); // `now` to interrupt potentially parked runnable
fsyncExecutor.awaitTermination(1, MINUTES);
}
diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java
index 2a63ab7cc4..3120eb0916 100644
--- a/src/java/org/apache/cassandra/journal/Journal.java
+++ b/src/java/org/apache/cassandra/journal/Journal.java
@@ -114,7 +114,7 @@ public class Journal<K, V> implements Shutdownable
final AtomicReference<State> state = new AtomicReference<>(State.UNINITIALIZED);
- // TODO (required): we do not need wait queues here, we can just wait on a signal on a segment while its byte buffer is being allocated
+ // TODO (expected): we do not need wait queues here, we can just wait on a signal on a segment while its byte buffer is being allocated
private final WaitQueue segmentPrepared = newWaitQueue();
private final WaitQueue allocatorThreadWaitQueue = newWaitQueue();
private final BooleanSupplier allocatorThreadWaitCondition = () -> (availableSegment == null);
@@ -256,6 +256,7 @@ public class Journal<K, V> implements Shutdownable
{
Invariants.require(state.compareAndSet(State.NORMAL, State.SHUTDOWN),
"Unexpected journal state while trying to shut down", state);
+ logger.debug("Shutting down " + allocator + " and awaiting termination");
allocator.shutdown();
wakeAllocator(); // Wake allocator to force it into shutdown
// TODO (expected): why are we awaitingTermination here when we have a separate method for it?
@@ -265,6 +266,7 @@ public class Journal<K, V> implements Shutdownable
compactor.awaitTermination(1, TimeUnit.MINUTES);
flusher.shutdown();
closeAllSegments();
+ logger.debug("Shutting down " + releaser + " and " + closer + " and awaiting termination");
releaser.shutdown();
closer.shutdown();
closer.awaitTermination(1, TimeUnit.MINUTES);
diff --git a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java
index 10fbfd6621..21f1297584 100644
--- a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java
@@ -94,13 +94,13 @@ public class AccordCacheMetrics
this.hits = Metrics.gauge(factory.createMetricName("Hits"), hitRate::totalHits);
this.misses = Metrics.gauge(factory.createMetricName("Misses"), hitRate::totalMisses);
this.requests = Metrics.gauge(factory.createMetricName("Requests"), hitRate::totalRequests);
- this.requestRate1m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(1));
- this.requestRate5m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(5));
+ this.requestRate1m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.ONE_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(1));
+ this.requestRate5m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIVE_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(5));
this.requestRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(15));
this.hitRate1m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.ONE_MINUTE + "HitRate"), () -> hitRate.hitRate(1));
this.hitRate5m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIVE_MINUTE + "HitRate"), () -> hitRate.hitRate(5));
this.hitRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "HitRate"), () -> hitRate.hitRate(15));
- this.hitRateAllTime = Metrics.gauge(factory.createMetricName("Misses"), hitRate::hitRateAllTime);
+ this.hitRateAllTime = Metrics.gauge(factory.createMetricName("HitRate"), hitRate::hitRateAllTime);
this.subTypeName = subTypeName;
}
diff --git a/src/java/org/apache/cassandra/metrics/AccordMetricUtils.java b/src/java/org/apache/cassandra/metrics/AccordMetricUtils.java
index 190664018e..d7b784fe8e 100644
--- a/src/java/org/apache/cassandra/metrics/AccordMetricUtils.java
+++ b/src/java/org/apache/cassandra/metrics/AccordMetricUtils.java
@@ -45,9 +45,9 @@ public class AccordMetricUtils
static <I, T> Gauge<T> fromAccordService(Function<IAccordService, I> initialStep, Function<I, T> finalStep, T ifNotSetup)
{
- if (AccordService.isSetup())
+ if (AccordService.isSetupOrStarting())
{
- IAccordService service = AccordService.instance();
+ IAccordService service = AccordService.unsafeInstance();
if (!service.isEnabled())
return () -> ifNotSetup;
I intermediate = initialStep.apply(service);
diff --git a/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java b/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java
index 11f5f970b3..313ed92bb2 100644
--- a/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java
@@ -138,7 +138,7 @@ public class AccordSystemMetrics
private AccordSystemMetrics()
{
- Invariants.expect(AccordService.isSetup());
+ Invariants.expect(AccordService.isSetupOrStarting());
DefaultNameFactory factory = new DefaultNameFactory(ACCORD_SYSTEM);
minEpoch = Metrics.gauge(factory.createMetricName(MIN_EPOCH), fromTopologyManager(TopologyManager::minEpoch));
maxEpoch = Metrics.gauge(factory.createMetricName(MAX_EPOCH), fromTopologyManager(TopologyManager::epoch));
@@ -164,13 +164,13 @@ public class AccordSystemMetrics
private synchronized void refreshHistograms()
{
- if (!AccordService.isSetup())
+ if (!AccordService.isSetupOrStarting())
{
snapshot = new Snapshot();
return;
}
- IAccordService service = AccordService.instance();
+ IAccordService service = AccordService.unsafeInstance();
if (!service.isEnabled())
{
snapshot = new Snapshot();
diff --git a/src/java/org/apache/cassandra/net/MessageDelivery.java b/src/java/org/apache/cassandra/net/MessageDelivery.java
index ccface2253..cb6c7190d7 100644
--- a/src/java/org/apache/cassandra/net/MessageDelivery.java
+++ b/src/java/org/apache/cassandra/net/MessageDelivery.java
@@ -174,7 +174,7 @@ public interface MessageDelivery
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
long retryDelay = backoff.computeWait(attempt + 1, NANOSECONDS);
- // TODO (required): we already have a separate retry predicate, retries should not be taken into consideration when retrying
+ // TODO (expected): we already have a separate retry predicate, retries should not be taken into consideration when retrying
if (retryDelay < 0)
{
onResult.result(attempt, null, new GivingUpException(attempt, errorMessage.apply(attempt, ResponseFailureReason.GiveUp, from, failure)));
diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java
index 8d55cab8f8..d4e4a48568 100644
--- a/src/java/org/apache/cassandra/net/Verb.java
+++ b/src/java/org/apache/cassandra/net/Verb.java
@@ -355,13 +355,13 @@ public enum Verb
ACCORD_GET_LATEST_DEPS_REQ (151, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(LatestDepsSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_LATEST_DEPS_RSP),
ACCORD_GET_MAX_CONFLICT_RSP (152, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.reply), AccordService::responseHandlerOrNoop ),
ACCORD_GET_MAX_CONFLICT_REQ (153, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP),
- ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ),
- ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ),
- ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
- ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
+ ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ),
+ ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ),
+ ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
+ ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_SYNC_NOTIFY_RSP (158, P2, writeTimeout, MISC, () -> accordEmbedded(EnumSerializer.simpleReply), RESPONSE_HANDLER),
- ACCORD_SYNC_NOTIFY_REQ (159, P2, writeTimeout, MISC, () -> accordEmbedded(Notification.serializer), () -> AccordSyncPropagator.verbHandler, ACCORD_SYNC_NOTIFY_RSP ),
+ ACCORD_SYNC_NOTIFY_REQ (159, P2, writeTimeout, MISC, () -> accordEmbedded(Notification.serializer), () -> AccordSyncPropagator.verbHandler, ACCORD_SYNC_NOTIFY_RSP ),
CONSENSUS_KEY_MIGRATION (160, P1, writeTimeout, MISC, () -> accordEmbedded(ConsensusKeyMigrationFinished.serializer),() -> ConsensusKeyMigrationState.consensusKeyMigrationFinishedHandler),
diff --git a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java
index ff3d5e6f9c..e07ae8cc1e 100644
--- a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java
+++ b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java
@@ -31,6 +31,7 @@ import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.repair.RepairCoordinator.NeighborsAndRanges;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Future;
@@ -44,13 +45,15 @@ public abstract class AbstractRepairTask implements RepairTask
protected final InetAddressAndPort broadcastAddressAndPort;
protected final RepairOption options;
protected final String keyspace;
+ protected final NeighborsAndRanges neighborsAndRanges;
- protected AbstractRepairTask(RepairCoordinator coordinator)
+ protected AbstractRepairTask(RepairCoordinator coordinator, NeighborsAndRanges neighborsAndRanges)
{
this.coordinator = Objects.requireNonNull(coordinator);
this.broadcastAddressAndPort = coordinator.ctx.broadcastAddressAndPort();
this.options = Objects.requireNonNull(coordinator.state.options);
this.keyspace = Objects.requireNonNull(coordinator.state.keyspace);
+ this.neighborsAndRanges = neighborsAndRanges;
}
private List<RepairSession> submitRepairSessions(TimeUUID parentSession,
@@ -71,6 +74,7 @@ public abstract class AbstractRepairTask implements RepairTask
excludedDeadNodes,
keyspace,
options.getParallelism(),
+ neighborsAndRanges.includesAllReplicas,
isIncremental,
options.isPullRepair(),
options.getPreviewKind(),
@@ -79,6 +83,7 @@ public abstract class AbstractRepairTask implements RepairTask
options.repairPaxos(),
options.dontPurgeTombstones(),
options.repairAccord(),
+ options.permitNoQuorum(),
executor,
validationScheduler,
cfnames);
diff --git a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java
index 956973c494..c752d2e39d 100644
--- a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java
+++ b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java
@@ -31,7 +31,6 @@ import org.apache.cassandra.utils.concurrent.Future;
public class IncrementalRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
- private final RepairCoordinator.NeighborsAndRanges neighborsAndRanges;
private final String[] cfnames;
protected IncrementalRepairTask(RepairCoordinator coordinator,
@@ -39,9 +38,8 @@ public class IncrementalRepairTask extends AbstractRepairTask
RepairCoordinator.NeighborsAndRanges neighborsAndRanges,
String[] cfnames)
{
- super(coordinator);
+ super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
- this.neighborsAndRanges = neighborsAndRanges;
this.cfnames = cfnames;
}
@@ -60,7 +58,7 @@ public class IncrementalRepairTask extends AbstractRepairTask
.add(broadcastAddressAndPort)
.build();
// Not necessary to include self for filtering. The common ranges only contains neighbhor node endpoints.
- List<CommonRange> allRanges = neighborsAndRanges.filterCommonRanges(keyspace, cfnames);
+ List<CommonRange> allRanges = neighborsAndRanges.filterCommonRanges(keyspace, cfnames).commonRanges;
CoordinatorSession coordinatorSession = coordinator.ctx.repair().consistent.coordinated.registerSession(parentSession, allParticipants, neighborsAndRanges.shouldExcludeDeadParticipants);
diff --git a/src/java/org/apache/cassandra/repair/NormalRepairTask.java b/src/java/org/apache/cassandra/repair/NormalRepairTask.java
index 05b721d17c..20b133c6eb 100644
--- a/src/java/org/apache/cassandra/repair/NormalRepairTask.java
+++ b/src/java/org/apache/cassandra/repair/NormalRepairTask.java
@@ -17,8 +17,6 @@
*/
package org.apache.cassandra.repair;
-import java.util.List;
-
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Future;
@@ -26,20 +24,15 @@ import org.apache.cassandra.utils.concurrent.Future;
public class NormalRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
- private final List<CommonRange> commonRanges;
- private final boolean excludedDeadNodes;
private final String[] cfnames;
protected NormalRepairTask(RepairCoordinator coordinator,
TimeUUID parentSession,
- List<CommonRange> commonRanges,
- boolean excludedDeadNodes,
+ RepairCoordinator.NeighborsAndRanges neighborsAndRanges,
String[] cfnames)
{
- super(coordinator);
+ super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
- this.commonRanges = commonRanges;
- this.excludedDeadNodes = excludedDeadNodes;
this.cfnames = cfnames;
}
@@ -52,6 +45,6 @@ public class NormalRepairTask extends AbstractRepairTask
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
- return runRepair(parentSession, false, executor, validationScheduler, commonRanges, excludedDeadNodes, cfnames);
+ return runRepair(parentSession, false, executor, validationScheduler, neighborsAndRanges.commonRanges, neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
}
}
diff --git a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java
index 6323d8e275..316c65ab54 100644
--- a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java
+++ b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java
@@ -42,17 +42,13 @@ import org.apache.cassandra.utils.concurrent.Future;
public class PreviewRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
- private final List<CommonRange> commonRanges;
- private final boolean excludedDeadNodes;
private final String[] cfnames;
private volatile String successMessage = name() + " completed successfully";
- protected PreviewRepairTask(RepairCoordinator coordinator, TimeUUID parentSession, List<CommonRange> commonRanges, boolean excludedDeadNodes, String[] cfnames)
+ protected PreviewRepairTask(RepairCoordinator coordinator, TimeUUID parentSession, RepairCoordinator.NeighborsAndRanges neighborsAndRanges, String[] cfnames)
{
- super(coordinator);
+ super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
- this.commonRanges = commonRanges;
- this.excludedDeadNodes = excludedDeadNodes;
this.cfnames = cfnames;
}
@@ -71,7 +67,7 @@ public class PreviewRepairTask extends AbstractRepairTask
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
- Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, validationScheduler, commonRanges, excludedDeadNodes, cfnames);
+ Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, validationScheduler, neighborsAndRanges.commonRanges, neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
return f.map(result -> {
if (result.hasFailed())
return result;
diff --git a/src/java/org/apache/cassandra/repair/RepairCoordinator.java b/src/java/org/apache/cassandra/repair/RepairCoordinator.java
index fa2036b02f..8a5045382b 100644
--- a/src/java/org/apache/cassandra/repair/RepairCoordinator.java
+++ b/src/java/org/apache/cassandra/repair/RepairCoordinator.java
@@ -394,6 +394,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
private NeighborsAndRanges getNeighborsAndRanges() throws RepairException
{
Set<InetAddressAndPort> allNeighbors = new HashSet<>();
+ Set<InetAddressAndPort> includeNeighbors = new HashSet<>();
List<CommonRange> commonRanges = new ArrayList<>();
//pre-calculate output of getLocalReplicas and pass it to getNeighbors to increase performance and prevent
@@ -403,10 +404,14 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
boolean isCMS = ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort());
for (Range<Token> range : state.options.getRanges())
{
- EndpointsForRange neighbors = ctx.repair().getNeighbors(state.keyspace, keyspaceLocalRanges, range,
- state.options.getDataCenters(),
- state.options.getHosts());
- if (neighbors.isEmpty())
+ EndpointsForRange allForRange = ctx.repair().getNeighbors(state.keyspace, keyspaceLocalRanges, range);
+ allNeighbors.addAll(allForRange.endpoints());
+
+ EndpointsForRange includeForRange = ctx.repair().filterNeighbors(allForRange, range,
+ state.options.getDataCenters(),
+ state.options.getHosts());
+
+ if (includeForRange.isEmpty())
{
if (state.options.ignoreUnreplicatedKeyspaces())
{
@@ -423,11 +428,11 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
throw RepairException.warn(String.format("Nothing to repair for %s in %s - aborting", range, state.keyspace));
}
}
- addRangeToNeighbors(commonRanges, range, neighbors);
- allNeighbors.addAll(neighbors.endpoints());
+ addRangeToNeighbors(commonRanges, range, includeForRange);
+ includeNeighbors.addAll(includeForRange.endpoints());
}
- if (allNeighbors.isEmpty())
+ if (includeNeighbors.isEmpty())
{
if (state.options.ignoreUnreplicatedKeyspaces())
{
@@ -447,11 +452,12 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
if (shouldExcludeDeadParticipants)
{
- Set<InetAddressAndPort> actualNeighbors = Sets.newHashSet(Iterables.filter(allNeighbors, ctx.failureDetector()::isAlive));
- shouldExcludeDeadParticipants = !allNeighbors.equals(actualNeighbors);
- allNeighbors = actualNeighbors;
+ Set<InetAddressAndPort> actualNeighbors = Sets.newHashSet(Iterables.filter(includeNeighbors, ctx.failureDetector()::isAlive));
+ shouldExcludeDeadParticipants = !includeNeighbors.equals(actualNeighbors);
+ if (shouldExcludeDeadParticipants) includeNeighbors = actualNeighbors;
+ else logger.info("{} all replicas {} considered up and healthy; clearing force flag for this job", state.id, includeNeighbors);
}
- return new NeighborsAndRanges(shouldExcludeDeadParticipants, allNeighbors, commonRanges);
+ return new NeighborsAndRanges(shouldExcludeDeadParticipants, includeNeighbors.containsAll(allNeighbors), includeNeighbors, commonRanges);
}
private void maybeStoreParentRepairStart(String[] cfnames)
@@ -496,7 +502,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
RepairTask task;
if (state.options.isPreview())
{
- task = new PreviewRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
+ task = new PreviewRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), cfnames);
}
else if (state.options.isIncremental())
{
@@ -504,7 +510,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
}
else
{
- task = new NormalRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
+ task = new NormalRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), cfnames);
}
ExecutorPlus executor = createExecutor();
@@ -635,12 +641,14 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
public static final class NeighborsAndRanges
{
final boolean shouldExcludeDeadParticipants;
+ public final boolean includesAllReplicas;
public final Set<InetAddressAndPort> participants;
public final List<CommonRange> commonRanges;
- public NeighborsAndRanges(boolean shouldExcludeDeadParticipants, Set<InetAddressAndPort> participants, List<CommonRange> commonRanges)
+ public NeighborsAndRanges(boolean shouldExcludeDeadParticipants, boolean includesAllReplicas, Set<InetAddressAndPort> participants, List<CommonRange> commonRanges)
{
this.shouldExcludeDeadParticipants = shouldExcludeDeadParticipants;
+ this.includesAllReplicas = includesAllReplicas;
this.participants = participants;
this.commonRanges = commonRanges;
}
@@ -650,11 +658,11 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
* and exludes ranges left without any participants
* When not in the force mode, no-op.
*/
- public List<CommonRange> filterCommonRanges(String keyspace, String[] tableNames)
+ public NeighborsAndRanges filterCommonRanges(String keyspace, String[] tableNames)
{
if (!shouldExcludeDeadParticipants)
{
- return commonRanges;
+ return this;
}
else
{
@@ -682,7 +690,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
}
}
Preconditions.checkState(!filtered.isEmpty(), "Not enough live endpoints for a repair");
- return filtered;
+ return new NeighborsAndRanges(shouldExcludeDeadParticipants, includesAllReplicas, participants, filtered);
}
}
}
diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java
index 4bebca17e4..5febd5be7b 100644
--- a/src/java/org/apache/cassandra/repair/RepairJob.java
+++ b/src/java/org/apache/cassandra/repair/RepairJob.java
@@ -39,6 +39,7 @@ import com.google.common.util.concurrent.FutureCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.local.durability.DurabilityService.SyncRemote;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
@@ -70,6 +71,9 @@ import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+import static accord.local.durability.DurabilityService.SyncRemote.All;
+import static accord.local.durability.DurabilityService.SyncRemote.NoRemote;
+import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static com.google.common.util.concurrent.Futures.getUnchecked;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosRepairEnabled;
import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME;
@@ -179,20 +183,18 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
if (doAccordRepair)
{
accordRepair = paxosRepair.flatMap(unused -> {
- boolean requireAllEndpoints;
- // If the session excluded dead nodes it's not eligible for migration and is not supposed to occur at ALL anyways
- if (session.excludedDeadNodes)
- requireAllEndpoints = false;
+ SyncRemote sync;
+ // If the session is doing a data repair (which flushes sstables if not incremental) we can do the barriers at QUORUM
+ if (session.excludedDeadNodes || !session.allReplicas || (session.repairData && !session.isIncremental))
+ {
+ sync = session.permitNoQuorum ? NoRemote : Quorum;
+ }
else
{
- // If the session is doing a data repair (which flushes sstables if not incremental) we can do the barriers at QUORUM
- if (session.repairData && !session.isIncremental)
- requireAllEndpoints = false;
- else
- requireAllEndpoints = true;
+ sync = All;
}
- logger.info("{} {}.{} starting accord repair, require all endpoints {}", session.previewKind.logPrefix(session.getId()), desc.keyspace, desc.columnFamily, requireAllEndpoints);
- AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, requireAllEndpoints, allEndpoints);
+ logger.info("{} {}.{} starting accord repair, sync {}", session.previewKind.logPrefix(session.getId()), desc.keyspace, desc.columnFamily, sync);
+ AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, sync, sync == All ? null : allEndpoints);
return repair.repair(taskExecutor).flatMap(accordRepairResult -> {
// Propagate the HLC discovered during Accord repair to Paxos so Paxos doesn't use ballots < Accord has already used
if (accordRepairResult.maxHlc != IAccordService.NO_HLC)
diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java
index 3285fec9db..d22c96bae3 100644
--- a/src/java/org/apache/cassandra/repair/RepairSession.java
+++ b/src/java/org/apache/cassandra/repair/RepairSession.java
@@ -119,12 +119,14 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
/** Range to repair */
public final boolean isIncremental;
+ public final boolean allReplicas;
public final PreviewKind previewKind;
public final boolean repairData;
- public final boolean repairPaxos; // TODO (now): rename to repairPaxosIfSupported
+ public final boolean repairPaxos;
public final boolean repairAccord;
public final boolean dontPurgeTombstones;
public final boolean excludedDeadNodes;
+ public final boolean permitNoQuorum;
private final AtomicBoolean isFailed = new AtomicBoolean(false);
@@ -161,6 +163,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
boolean excludedDeadNodes,
String keyspace,
RepairParallelism parallelismDegree,
+ boolean allReplicas,
boolean isIncremental,
boolean pullRepair,
PreviewKind previewKind,
@@ -168,7 +171,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
boolean repairData,
boolean repairPaxos,
boolean dontPurgeTombstones,
- boolean repairAccord,
+ boolean repairAccord, boolean permitNoQuorum,
String... cfnames)
{
this.ctx = ctx;
@@ -176,12 +179,14 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
this.repairData = repairData;
this.repairPaxos = repairPaxos;
this.repairAccord = repairAccord;
+ this.permitNoQuorum = permitNoQuorum;
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";
this.state = new SessionState(ctx, parentRepairSession, keyspace, cfnames, commonRange);
this.parallelismDegree = parallelismDegree;
this.isIncremental = isIncremental;
this.previewKind = previewKind;
this.pullRepair = pullRepair;
+ this.allReplicas = allReplicas && !commonRange.hasSkippedReplicas;
this.optimiseStreams = optimiseStreams;
this.dontPurgeTombstones = dontPurgeTombstones;
this.taskExecutor = new SafeExecutor(createExecutor(ctx));
diff --git a/src/java/org/apache/cassandra/repair/SharedContext.java b/src/java/org/apache/cassandra/repair/SharedContext.java
index 790fe94886..cffa692bd1 100644
--- a/src/java/org/apache/cassandra/repair/SharedContext.java
+++ b/src/java/org/apache/cassandra/repair/SharedContext.java
@@ -52,7 +52,7 @@ import org.apache.cassandra.utils.TimeUUID;
*
* See {@link Global#instance} for the main production path
*/
-// TODO (required, clarity): move under Util since this is a class with shared logic
+// TODO (expected): move under Util since this is a class with shared logic
public interface SharedContext
{
InetAddressAndPort broadcastAddressAndPort();
diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java
index fe0f566c46..ef64f24066 100644
--- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java
+++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairState.java
@@ -311,7 +311,7 @@ class PreviewRepairedState extends AutoRepairState
public RepairCoordinator getRepairRunnable(String keyspace, List<String> tables, Set<Range<Token>> ranges, boolean primaryRangeOnly)
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false,
- AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false);
+ AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false, false);
option.getColumnFamilies().addAll(tables);
@@ -331,7 +331,7 @@ class IncrementalRepairState extends AutoRepairState
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, true, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges,
- false, false, PreviewKind.NONE, true, true, true, false, false, false);
+ false, false, PreviewKind.NONE, true, true, true, false, false, false, false);
option.getColumnFamilies().addAll(filterOutUnsafeTables(keyspace, tables));
@@ -376,7 +376,7 @@ class FullRepairState extends AutoRepairState
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges,
- false, false, PreviewKind.NONE, true, true, true, false, false, false);
+ false, false, PreviewKind.NONE, true, true, true, false, false, false, false);
option.getColumnFamilies().addAll(tables);
diff --git a/src/java/org/apache/cassandra/repair/messages/RepairOption.java b/src/java/org/apache/cassandra/repair/messages/RepairOption.java
index 3c8a260164..4cbff03657 100644
--- a/src/java/org/apache/cassandra/repair/messages/RepairOption.java
+++ b/src/java/org/apache/cassandra/repair/messages/RepairOption.java
@@ -63,6 +63,7 @@ public class RepairOption
public static final String REPAIR_PAXOS_KEY = "repairPaxos";
public static final String NO_TOMBSTONE_PURGING = "nopurge";
public static final String REPAIR_ACCORD_KEY = "repairAccord";
+ public static final String PERMIT_NO_QUORUM_KEY = "permitNoQuorum";
// we don't want to push nodes too much for repair
public static final int MAX_JOB_THREADS = 4;
@@ -209,6 +210,7 @@ public class RepairOption
logger.info("Overriding and disabling Accord repair because Accord is not enabled");
repairAccord = false;
}
+ boolean permitNoQuorum = Boolean.parseBoolean(options.get(PERMIT_NO_QUORUM_KEY));
if (previewKind != PreviewKind.NONE)
{
@@ -233,7 +235,7 @@ public class RepairOption
boolean asymmetricSyncing = Boolean.parseBoolean(options.get(OPTIMISE_STREAMS_KEY));
- RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, pullRepair, force, previewKind, asymmetricSyncing, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord);
+ RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, pullRepair, force, previewKind, asymmetricSyncing, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord, permitNoQuorum);
// data centers
String dataCentersStr = options.get(DATACENTERS_KEY);
@@ -316,13 +318,14 @@ public class RepairOption
private final boolean repairPaxos;
private final boolean dontPurgeTombstones;
private final boolean repairAccord;
+ private final boolean permitNoQuorum; // has an effect only for Accord at present
private final Collection<String> columnFamilies = new HashSet<>();
private final Collection<String> dataCenters = new HashSet<>();
private final Collection<String> hosts = new HashSet<>();
private final Collection<Range<Token>> ranges = new HashSet<>();
- public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean pullRepair, boolean forceRepair, PreviewKind previewKind, boolean optimiseStreams, boolean ignoreUnreplicatedKeyspaces, boolean repairData, boolean repairPaxos, boolean dontPurgeTombstones, boolean repairAccord)
+ public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean pullRepair, boolean forceRepair, PreviewKind previewKind, boolean optimiseStreams, boolean ignoreUnreplicatedKeyspaces, boolean repairData, boolean repairPaxos, boolean dontPurgeTombstones, boolean repairAccord, boolean permitNoQuorum)
{
checkArgument(repairData || repairAccord || repairPaxos, "Repair needs to repair at least one of data, Paxos, or Accord");
this.parallelism = parallelism;
@@ -340,6 +343,7 @@ public class RepairOption
this.repairPaxos = repairPaxos;
this.dontPurgeTombstones = dontPurgeTombstones;
this.repairAccord = repairAccord;
+ this.permitNoQuorum = permitNoQuorum;
}
public RepairParallelism getParallelism()
@@ -460,6 +464,11 @@ public class RepairOption
return repairAccord;
}
+ public boolean permitNoQuorum()
+ {
+ return repairAccord;
+ }
+
public boolean dontPurgeTombstones()
{
return dontPurgeTombstones;
@@ -514,6 +523,7 @@ public class RepairOption
options.put(REPAIR_PAXOS_KEY, Boolean.toString(repairPaxos));
options.put(NO_TOMBSTONE_PURGING, Boolean.toString(dontPurgeTombstones));
options.put(REPAIR_ACCORD_KEY, Boolean.toString(repairAccord));
+ options.put(PERMIT_NO_QUORUM_KEY, Boolean.toString(permitNoQuorum));
return options;
}
}
diff --git a/src/java/org/apache/cassandra/repair/state/CoordinatorState.java b/src/java/org/apache/cassandra/repair/state/CoordinatorState.java
index 70dcb5de7f..70a9885501 100644
--- a/src/java/org/apache/cassandra/repair/state/CoordinatorState.java
+++ b/src/java/org/apache/cassandra/repair/state/CoordinatorState.java
@@ -132,7 +132,7 @@ public class CoordinatorState extends AbstractState<CoordinatorState.State, Time
{
if (neighborsAndRanges == null)
return null;
- return neighborsAndRanges.filterCommonRanges(keyspace, getColumnFamilyNames());
+ return neighborsAndRanges.filterCommonRanges(keyspace, getColumnFamilyNames()).commonRanges;
}
@Override
diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java
index 27e50157a3..a6641eccd5 100644
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@ -458,6 +458,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
boolean excludedDeadNodes,
String keyspace,
RepairParallelism parallelismDegree,
+ boolean allReplicas,
boolean isIncremental,
boolean pullRepair,
PreviewKind previewKind,
@@ -466,6 +467,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
boolean repairPaxos,
boolean dontPurgeTombstones,
boolean repairAccord,
+ boolean permitNoQuorum,
ExecutorPlus executor,
Scheduler validationScheduler,
String... cfnames)
@@ -481,9 +483,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
final RepairSession session = new RepairSession(ctx, validationScheduler, parentRepairSession,
range, excludedDeadNodes, keyspace,
- parallelismDegree, isIncremental, pullRepair,
+ parallelismDegree, allReplicas, isIncremental, pullRepair,
previewKind, optimiseStreams, repairData, repairPaxos,
- dontPurgeTombstones, repairAccord, cfnames);
+ dontPurgeTombstones, repairAccord, permitNoQuorum, cfnames);
repairs.getIfPresent(parentRepairSession).register(session.state);
sessions.put(session.getId(), session);
@@ -567,6 +569,19 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
public EndpointsForRange getNeighbors(String keyspaceName, Iterable<Range<Token>> keyspaceLocalRanges,
Range<Token> toRepair, Collection<String> dataCenters,
Collection<String> hosts)
+ {
+ return filterNeighbors(getNeighbors(keyspaceName, keyspaceLocalRanges, toRepair), toRepair, dataCenters, hosts);
+ }
+
+ /**
+ * Return all of the neighbors with whom we share the provided range.
+ *
+ * @param keyspaceName keyspace to repair
+ * @param keyspaceLocalRanges local-range for given keyspaceName
+ * @param toRepair token to repair
+ * @return neighbors with whom we share the provided range
+ */
+ public EndpointsForRange getNeighbors(String keyspaceName, Iterable<Range<Token>> keyspaceLocalRanges, Range<Token> toRepair)
{
StorageService ss = StorageService.instance;
EndpointsByRange replicaSets = ss.getRangeToAddressMap(keyspaceName);
@@ -589,7 +604,22 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
return EndpointsForRange.empty(toRepair);
// same as withoutSelf(), but done this way for testing
- EndpointsForRange neighbors = replicaSets.get(rangeSuperSet).filter(r -> !ctx.broadcastAddressAndPort().equals(r.endpoint()));
+ return replicaSets.get(rangeSuperSet).filter(r -> !ctx.broadcastAddressAndPort().equals(r.endpoint()));
+ }
+
+
+ /**
+ * Return all of the neighbors in the listed data center or host lists
+ *
+ * @param toRepair token to repair
+ * @param dataCenters the data centers to involve in the repair
+ * @return neighbors with whom we share the provided range
+ */
+ public EndpointsForRange filterNeighbors(EndpointsForRange neighbors, Range<Token> toRepair, Collection<String> dataCenters,
+ Collection<String> hosts)
+ {
+ if (neighbors.isEmpty())
+ return neighbors;
ClusterMetadata metadata = ClusterMetadata.current();
if (dataCenters != null && !dataCenters.isEmpty())
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 973dfbfe92..22b8810f48 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -74,6 +74,7 @@ import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Locator;
+import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.tcm.CMSOperations;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.RegistrationStatus;
@@ -91,6 +92,7 @@ import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.membership.NodeState;
+import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
@@ -343,11 +345,12 @@ public class CassandraDaemon
PaxosState.initializeTrackers();
// replay the log if necessary
- // TODO samt - when restarting a previously running instance, this needs to happen after reconstructing schema
- // from the cluster metadata log or all mutations will throw IncompatibleSchemaException on deserialisation
try
{
CommitLog.instance.recoverSegmentsOnDisk();
+ NodeId self = ClusterMetadata.current().myNodeId();
+ if (self != null)
+ AccordService.localStartup(self);
}
catch (IOException e)
{
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 5f41f57c5c..a5703fe601 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -847,7 +847,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
Gossiper.waitToSettle();
NodeId self = Register.maybeRegister();
- AccordService.startup(self);
+ if (!AccordService.isSetupOrStarting())
+ AccordService.localStartup(self);
+ AccordService.distributedStartup();
+
RegistrationStatus.instance.onRegistration();
Startup.maybeExecuteStartupTransformation(self);
@@ -3165,7 +3168,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
true, // repairData
false, // repairPaxos
true, // dontPurgeTombstones
- false // repairAccord
+ false, // repairAccord
+ false // permit no quorum
);
return new RepairCoordinator(this, cmd, options, keyspace);
@@ -3846,8 +3850,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
transientMode = Optional.of(Mode.DRAINING);
}
- if (AccordService.isSetup())
- AccordService.instance().markShuttingDown();
+ if (AccordService.isSetupOrStarting())
+ AccordService.unsafeInstance().markShuttingDown();
// In-progress writes originating here could generate hints to be written,
// which is currently scheduled on the mutation stage. So shut down MessagingService
@@ -3863,14 +3867,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
logger.error("Messaging service timed out shutting down", t);
}
- if (AccordService.isSetup())
+ if (AccordService.isSetupOrStarting())
{
logger.info("Flushing Accord caches");
- if (!AccordService.instance().flushCaches().awaitUninterruptibly(1, MINUTES))
+ if (!AccordService.unsafeInstance().flushCaches().awaitUninterruptibly(1, MINUTES))
logger.error("Could not flush Accord caches promptly");
if (AccordColumnFamilyStores.commandsForKey != null)
AccordColumnFamilyStores.commandsForKey.forceBlockingFlush(INTERNALLY_FORCED);
- AccordService.instance().shutdownAndWait(1, MINUTES);
+ AccordService.unsafeInstance().shutdownAndWait(1, MINUTES);
}
// ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it.
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java
index c17ccc879c..fc794f6dec 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
@@ -60,6 +61,7 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.metrics.LogLinearHistogram;
import org.apache.cassandra.metrics.ShardedHitRate;
+import org.apache.cassandra.service.accord.AccordCache.Adapter.Shrink;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import org.apache.cassandra.service.accord.events.CacheEvents;
@@ -100,11 +102,14 @@ public class AccordCache implements CacheSize
public interface Adapter<K, V, S>
{
+ enum Shrink { EVICT, DONE, PERFORM_WITHOUT_LOCK }
+
@Nullable V load(AccordCommandStore commandStore, K key);
@Nullable Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk);
default boolean canSave(@Nullable V value, @Nullable Object shrunk) { return true; }
// a result of null means we can immediately evict, without saving
@Nullable V quickShrink(V value);
+ Shrink decideFullShrink(K key, V value);
// a result of null means we cannot shrink, and should save/evict as appropriate
@Nullable Object fullShrink(K key, V value);
@Nullable V inflate(AccordCommandStore commandStore, K key, Object shrunk);
@@ -149,6 +154,7 @@ public class AccordCache implements CacheSize
private long bytesCached;
private int noEvictGeneration;
private boolean shrinkingOn = true;
+ private boolean tryShrinkOrEvict;
public AccordCache(AccordCacheEntry.SaveExecutor saveExecutor, long maxSizeInBytes)
{
@@ -156,11 +162,12 @@ public class AccordCache implements CacheSize
this.maxSizeInBytes = maxSizeInBytes;
}
+ // note: only affects current contents after lock is released
@Override
public void setCapacity(long sizeInBytes)
{
maxSizeInBytes = sizeInBytes;
- maybeShrinkOrEvictSomeNodes();
+ tryShrinkOrEvict = true;
}
public void setShrinkingOn(boolean shrinkingOn)
@@ -206,31 +213,56 @@ public class AccordCache implements CacheSize
* Roughly respects LRU semantics when evicting. Might consider prioritising keeping MODIFIED nodes around
* for longer to maximise the chances of hitting system tables fewer times (or not at all).
*/
- private void maybeShrinkOrEvictSomeNodes()
+ void tryShrinkOrEvict(Lock lock)
{
+ if (!tryShrinkOrEvict)
+ return;
+
while (bytesCached > maxSizeInBytes && !evictQueue.isEmpty())
{
AccordCacheEntry<?, ?> node = evictQueue.peek();
- shrinkOrEvict(node);
+ shrinkOrEvict(lock, node);
}
+ tryShrinkOrEvict = false;
}
@VisibleForTesting
- private <K, V> void shrinkOrEvict(AccordCacheEntry<K, V> node)
+ private <K, V> void shrinkOrEvict(Lock lock, AccordCacheEntry<K, V> node)
{
require(node.references() == 0);
- // TODO (desired): special-case evict queue with 1 element as no point shrinking first
- if (shrinkingOn && node.tryShrink())
+ Shrink shrink = shrinkingOn ? node.tryShrink() : Shrink.EVICT;
+ if (shrink == Shrink.EVICT)
{
- IntrusiveLinkedList<AccordCacheEntry<?,?>> queue;
- queue = node.isNoEvict() ? noEvictQueue : evictQueue;
- node.unlink();
- queue.addLast(node);
+ tryEvict(node);
}
else
{
- tryEvict(node);
+ IntrusiveLinkedList<AccordCacheEntry<?,?>> queue;
+ queue = node.isNoEvict() ? noEvictQueue : evictQueue;
+ node.unlink();
+ if (shrink == Shrink.DONE)
+ {
+ queue.addLast(node);
+ }
+ else
+ {
+ K key = node.key();
+ V cur = node.tryGetExclusive();
+ Object upd = null;
+ lock.unlock();
+ try
+ {
+ upd = node.owner.parent().adapter().fullShrink(key, cur);
+ }
+ finally
+ {
+ //noinspection LockAcquiredButNotSafelyReleased
+ lock.lock();
+ node.tryApplyShrink(cur, upd);
+ queue.addLast(node);
+ }
+ }
}
}
@@ -463,7 +495,7 @@ public class AccordCache implements CacheSize
node.initSize(parent());
++size;
node.notifyListeners(Listener::onAdd);
- maybeShrinkOrEvictSomeNodes();
+ tryShrinkOrEvict = true;
return node;
}
@@ -557,7 +589,7 @@ public class AccordCache implements CacheSize
}
}
- maybeShrinkOrEvictSomeNodes();
+ tryShrinkOrEvict = true;
}
AccordCacheEntry<K, ?> remove(K key)
@@ -1009,6 +1041,12 @@ public class AccordCache implements CacheSize
return quickShrink.apply(value);
}
+ @Override
+ public Shrink decideFullShrink(K key, V value)
+ {
+ return Shrink.DONE;
+ }
+
@Override
public Object fullShrink(K key, V value)
{
@@ -1087,6 +1125,7 @@ public class AccordCache implements CacheSize
@Override public V load(AccordCommandStore commandStore, K key) { return null; }
@Override public Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk) { return null; }
@Override public V quickShrink(V value) { return null; }
+ @Override public Shrink decideFullShrink(K key, V value) { return Shrink.DONE; }
@Override public Object fullShrink(K key, V value) { return null; }
@Override public V inflate(AccordCommandStore commandStore, K key, Object shrunk) { return null; }
@Override public long estimateHeapSize(V value) { return 0; }
@@ -1129,6 +1168,18 @@ public class AccordCache implements CacheSize
return value;
}
+ @Override
+ public Shrink decideFullShrink(RoutingKey key, CommandsForKey value)
+ {
+ if (value.isEmpty() || value.isLoadingPruned())
+ return Shrink.EVICT;
+
+ if (value.size() < 64)
+ return Shrink.DONE;
+
+ return Shrink.PERFORM_WITHOUT_LOCK;
+ }
+
@Override
public Object fullShrink(RoutingKey key, CommandsForKey value)
{
@@ -1249,6 +1300,19 @@ public class AccordCache implements CacheSize
}
}
+ @Override
+ public Shrink decideFullShrink(TxnId txnId, Command value)
+ {
+ if (txnId.is(Txn.Kind.EphemeralRead))
+ Invariants.expect(value.saveStatus().compareTo(SaveStatus.ReadyToExecute) < 0);
+
+ // TODO (expected): improve heuristics and consider transaction size
+ if (value.partialDeps() == null || value.partialDeps().txnIds().size() < 64)
+ return Shrink.DONE;
+
+ return Shrink.PERFORM_WITHOUT_LOCK;
+ }
+
@Override
public @Nullable Command inflate(AccordCommandStore commandStore, TxnId key, Object serialized)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java
index 9a39874083..02ee11c3db 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java
@@ -34,6 +34,7 @@ import accord.utils.IntrusiveLinkedListNode;
import accord.utils.Invariants;
import accord.utils.async.Cancellable;
import org.apache.cassandra.service.accord.AccordCache.Adapter;
+import org.apache.cassandra.service.accord.AccordCache.Adapter.Shrink;
import org.apache.cassandra.utils.ObjectSizes;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED;
@@ -462,16 +463,26 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
state = null;
}
- boolean tryShrink()
+ Shrink tryShrink()
{
if (!isLoaded())
- return false;
+ return Shrink.EVICT;
AccordCache.Type<K, V, ?> parent = owner.parent();
- if (!tryShrink(key, parent.adapter()))
- return false;
- updateSize(parent);
- return true;
+ Adapter<K, V, ?> adapter = parent.adapter();
+ if (isShrunk() || state == null)
+ return Shrink.EVICT;
+
+ V cur = (V)unwrap();
+ Shrink shrink = adapter.decideFullShrink(key, cur);
+ if (shrink == Shrink.PERFORM_WITHOUT_LOCK)
+ return Shrink.PERFORM_WITHOUT_LOCK;
+
+ Object upd = adapter.fullShrink(key, cur);
+ if (upd == null || upd == cur)
+ return Shrink.EVICT;
+ applyShrink(parent, cur, upd);
+ return Shrink.DONE;
}
V tryGetFull()
@@ -584,20 +595,18 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
return ((FailedToSave)state).cause;
}
- private boolean tryShrink(K key, Adapter<K, V, ?> adapter)
+ void tryApplyShrink(Object cur, Object upd)
{
- if (isShrunk() || state == null)
- return false;
-
- Object cur = unwrap();
- Object upd = adapter.fullShrink(key, (V)cur);
- if (upd == null || upd == cur)
- return false;
+ if (isLoaded() && unwrap() == cur && upd != cur && upd != null)
+ applyShrink(owner.parent(), cur, upd);
+ }
+ private void applyShrink(AccordCache.Type<K, V, ?> parent, Object cur, Object upd)
+ {
if (isNested()) ((Nested)this.state).state = upd;
else this.state = upd;
status |= SHRUNK;
- return true;
+ updateSize(parent);
}
private void inflate(AccordCommandStore commandStore, K key, Adapter<K, V, ?> adapter)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 99b763de21..d88314c320 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -142,6 +142,7 @@ public class AccordCommandStore extends CommandStore
@Override
public void close()
{
+ global().tryShrinkOrEvict(lock);
lock.unlock();
}
}
@@ -208,10 +209,14 @@ public class AccordCommandStore extends CommandStore
ranges = update.newRangesForEpoch;
Invariants.require(ranges != null, "CommandStore %d created with no ranges", id);
}
+
tableId = (TableId)ranges.all().stream().map(r -> r.start().prefix()).reduce((a, b) -> {
Invariants.require(a.equals(b), "CommandStore created with multiple distinct TableId (%s and %s)", a, b);
return a;
}).orElseThrow(() -> Invariants.illegalState("CommandStore %d created with no ranges", id));
+
+ if (AccordService.isStarted())
+ progressLog.unsafeStart();
}
static Factory factory(IntFunction<AccordExecutor> executorFactory)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
index 6d06fd47f2..ddc66f36e0 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
@@ -32,6 +32,7 @@ import accord.local.SequentialAsyncExecutor;
import accord.local.ShardDistributor;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
+import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.AccordSpec.QueueShardModel;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory;
@@ -66,6 +67,14 @@ public class AccordCommandStores extends CommandStores implements CacheSize
maxQueuedRangeLoads = DatabaseDescriptor.getAccordMaxQueuedRangeLoadCount();
shrinkingOn = DatabaseDescriptor.getAccordCacheShrinkingOn();
refreshCapacities();
+ ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> {
+ for (AccordExecutor executor : executors)
+ {
+ executor.executeDirectlyWithLock(() -> {
+ executor.cacheExclusive().processNoEvictQueue();
+ });
+ }
+ }, 1L, 1L, TimeUnit.SECONDS);
}
static Factory factory()
@@ -204,6 +213,9 @@ public class AccordCommandStores extends CommandStores implements CacheSize
for (AccordExecutor executor : executors)
{
executor.shutdown();
+ }
+ for (AccordExecutor executor : executors)
+ {
try
{
executor.awaitTermination(1, TimeUnit.MINUTES);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java
index dc9757ec45..438bca469f 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java
@@ -96,7 +96,16 @@ public class AccordDataStore implements DataStore
default: throw new UnhandledEnum(kind);
case Image:
{
- AccordFetchCoordinator coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
+ AccordFetchCoordinator coordinator;
+ try
+ {
+ coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
+ }
+ catch (Throwable t)
+ {
+ return new FetchResult.Failure(t);
+ }
+
coordinator.start();
return coordinator.result();
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java
index c9731d101a..bca114a0c6 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java
@@ -24,7 +24,6 @@ import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
@@ -66,7 +65,6 @@ import accord.utils.async.Cancellable;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.DebuggableTask;
import org.apache.cassandra.concurrent.DebuggableTask.DebuggableTaskRunner;
-import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.metrics.AccordExecutorMetrics;
@@ -130,6 +128,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
public void close()
{
executor.beforeUnlock();
+ global.tryShrinkOrEvict(lock);
lock.unlock();
}
}
@@ -235,9 +234,6 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
this.elapsedRunning = AccordExecutorMetrics.INSTANCE.elapsedRunning.forShard(histogramsShard);
this.keys = AccordExecutorMetrics.INSTANCE.keys.forShard(histogramsShard);
this.replicaMetrics = new AccordReplicaMetrics.Shard(histogramsShard);
- ScheduledExecutors.scheduledFastTasks.scheduleAtFixedRate(() -> {
- executeDirectlyWithLock(cache::processNoEvictQueue);
- }, 1L, 1L, TimeUnit.SECONDS);
}
public int executorId()
@@ -493,7 +489,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
catch (Throwable t)
{
- agent.onUncaughtException(t);
+ agent.onException(t);
}
}
@@ -663,6 +659,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
if (waitingForCompletion != null && waitingForCompletion.peek().maybeNotify - position >= 0)
maybeNotifyWaitingForCompletion();
+
+ cache.tryShrinkOrEvict(lock);
}
}
@@ -745,7 +743,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return;
try { task.failExclusive(fail); }
- catch (Throwable t) { agent.onUncaughtException(t); }
+ catch (Throwable t) { agent.onException(t); }
finally
{
task.unqueueIfQueued();
@@ -1197,7 +1195,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return false;
try { run.run(); }
- catch (Throwable t) { agent.onUncaughtException(t); }
+ catch (Throwable t) { agent.onException(t); }
finally
{
if (owner == null)
@@ -1322,7 +1320,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
queue.remove(this);
completeTaskExclusive(this);
try { fail(new CancellationException()); }
- catch (Throwable t) { agent.onUncaughtException(t); }
+ catch (Throwable t) { agent.onException(t); }
}
}
@@ -1373,7 +1371,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
{
if (result != null)
result.tryFailure(t);
- agent.onUncaughtException(t);
+ agent.onException(t);
}
@Override
@@ -1606,7 +1604,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
catch (Throwable t)
{
- agent.onUncaughtException(t);
+ agent.onException(t);
return;
}
}
@@ -1621,7 +1619,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
catch (Throwable t)
{
fail.addSuppressed(t);
- agent.onUncaughtException(fail);
+ agent.onException(fail);
}
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
index a23e47787f..f48deb9863 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
@@ -204,7 +204,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
pauseExclusive();
exitLockExclusive();
- try { agent.onUncaughtException(t); }
+ try { agent.onException(t); }
catch (Throwable t2) { }
}
finally
@@ -271,13 +271,13 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
catch (Throwable t2) { t.addSuppressed(t2); }
try { completeTaskExclusive(task); }
catch (Throwable t2) { t.addSuppressed(t2); }
- try { agent.onUncaughtException(t); }
+ try { agent.onException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }
task = null;
}
else
{
- try { agent.onUncaughtException(t); }
+ try { agent.onException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }
}
if (isHeldByExecutor)
@@ -302,7 +302,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
try
{
t2.addSuppressed(t);
- agent.onUncaughtException(t2);
+ agent.onException(t2);
}
catch (Throwable t3)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
index 84880f2e73..1006efb825 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
@@ -50,6 +50,7 @@ import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
+import accord.topology.TopologyException;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
@@ -78,6 +79,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.TimeUUID;
+import static accord.primitives.Routables.Slice.Minimal;
import static org.apache.cassandra.utils.CollectionSerializers.deserializeMap;
import static org.apache.cassandra.utils.CollectionSerializers.serializeMap;
import static org.apache.cassandra.utils.CollectionSerializers.serializedMapSize;
@@ -219,7 +221,6 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
Invariants.nonNull(future);
Invariants.require(this.future == null, "future was not null: %s", this.future);
this.future = future;
- logger.info("StreamFuture for plan {} received for bootstrap of {} from {}", planId, range, from);
maybeListen();
}
@@ -229,6 +230,8 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
if (range == null || future == null)
return;
+ logger.info("StreamFuture for plan {} received for bootstrap of {} from {}", planId, range, from);
+
Invariants.nonNull(from);
future.addCallback((state, fail) -> {
if (fail == null)
@@ -328,10 +331,10 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
}
@Override
- public Read slice(Ranges ranges) { return new StreamingRead(to, this.ranges.slice(ranges)); }
+ public Read slice(Ranges ranges) { return new StreamingRead(to, this.ranges.slice(ranges, Minimal)); }
@Override
- public Read intersecting(Participants<?> participants) { return new StreamingRead(to, this.ranges.slice(ranges)); }
+ public Read intersecting(Participants<?> participants) { return new StreamingRead(to, this.ranges.slice(ranges, Minimal)); }
@Override
public Read merge(Read other) { throw new UnsupportedOperationException(); }
@@ -391,7 +394,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
private final Map<TimeUUID, IncomingStream> streams = new HashMap<>();
- public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore)
+ public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore) throws TopologyException
{
super(node, node.someSequentialExecutor(), ranges, syncPoint, fetchRanges, commandStore);
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index 6633a65618..92c3629abe 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -798,6 +798,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public static @Nullable ByteBuffer asSerializedChange(Command before, Command after, Version userVersion) throws IOException
{
+ // TODO (expected): reusable buffer to build, or pre-size
try (DataOutputBuffer out = new DataOutputBuffer())
{
Writer writer = Writer.make(before, after);
@@ -1149,7 +1150,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion);
break;
case PARTIAL_DEPS:
- // TODO (required): this optimisation will be easily disabled;
+ // TODO (expected): this optimisation will be easily disabled;
// should either operate natively on ByteBuffer
// or else use some explicit API for copying bytes while skipping
if (deserializeDeps || !(in instanceof DataInputBuffer))
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
index 5d2a565932..69ff7fac10 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
@@ -36,7 +36,6 @@ import org.apache.cassandra.service.accord.serializers.Version;
import static accord.local.CommandStores.RangesForEpoch;
-// TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger
public class AccordJournalValueSerializers
{
public interface FlyweightImage
diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
index ab4d9c98f6..c302e684fb 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
@@ -25,8 +25,6 @@ import java.util.Set;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import accord.api.AsyncExecutor;
import accord.api.MessageSink;
@@ -48,7 +46,6 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.TimeoutStrategy;
-import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.utils.Clock;
import static accord.messages.MessageType.StandardMessage.*;
@@ -59,8 +56,6 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.slowR
public class AccordMessageSink implements MessageSink
{
- private static final Logger logger = LoggerFactory.getLogger(AccordMessageSink.class);
-
public enum AccordMessageType implements MessageType
{
INTEROP_READ_REQ(Verb.ACCORD_INTEROP_READ_REQ),
@@ -150,22 +145,20 @@ public class AccordMessageSink implements MessageSink
}
}
- private final AccordAgent agent;
private final MessageDelivery messaging;
private final AccordEndpointMapper endpointMapper;
private final RequestCallbacks callbacks;
- public AccordMessageSink(AccordAgent agent, MessageDelivery messaging, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
+ public AccordMessageSink(MessageDelivery messaging, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
{
- this.agent = agent;
this.messaging = messaging;
this.endpointMapper = endpointMapper;
this.callbacks = callbacks;
}
- public AccordMessageSink(AccordAgent agent, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
+ public AccordMessageSink(AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
{
- this(agent, MessagingService.instance(), endpointMapper, callbacks);
+ this(MessagingService.instance(), endpointMapper, callbacks);
}
@Override
@@ -177,7 +170,7 @@ public class AccordMessageSink implements MessageSink
InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(to, message);
if (endpoint == null)
return;
- logger.trace("Sending {} {} to {}", verb, message.payload, endpoint);
+
messaging.send(message, endpoint);
}
@@ -222,7 +215,6 @@ public class AccordMessageSink implements MessageSink
return null;
}
- logger.trace("Sending {} {} to {}", verb, message.payload, endpoint);
Cancellable cancellable = callbacks.registerAt(message.id(), executor, callback, to, nowNanos, slowAtNanos, expiresAtNanos, NANOSECONDS);
messaging.send(message, endpoint);
return cancellable;
@@ -240,7 +232,6 @@ public class AccordMessageSink implements MessageSink
if (endpoint == null)
return;
- logger.trace("Replying {} {} to {}", message.verb(), message.payload, endpoint);
messaging.send(message, endpoint);
}
@@ -252,7 +243,7 @@ public class AccordMessageSink implements MessageSink
InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(replyingTo, message);
if (endpoint == null)
return;
- logger.trace("Replying with failure {} {} to {}", message.verb(), message.payload, endpoint);
+
messaging.send(message, endpoint);
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java
index e75a74e1f3..e003fe6c03 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java
@@ -54,12 +54,6 @@ class AccordResponseVerbHandler<T extends Reply> implements IVerbHandler<T>
@Override
public void doVerb(Message message)
{
- if (!AccordService.instance().shouldAcceptMessages())
- {
- dropping.debug(message.verb(), message.from());
- return;
- }
-
Node.Id from = endpointMapper.mappedIdOrNull(message.from(), message);
if (from == null)
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordResult.java b/src/java/org/apache/cassandra/service/accord/AccordResult.java
index e0500c48fa..5e9cb88ab2 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordResult.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordResult.java
@@ -32,10 +32,10 @@ import accord.coordinate.CoordinationFailed;
import accord.coordinate.Exhausted;
import accord.coordinate.Preempted;
import accord.coordinate.Timeout;
-import accord.coordinate.TopologyMismatch;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.TxnId;
+import accord.topology.TopologyMismatch;
import accord.utils.UnhandledEnum;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.ReadFailureException;
@@ -152,26 +152,6 @@ public class AccordResult<V> extends AsyncFuture<V> implements BiConsumer<V, Thr
{
report = bookkeeping.newExhausted(txnId, keysOrRanges);
}
- else if (isTxnRequest && coordinationFailed instanceof TopologyMismatch)
- {
- // Excluding bugs topology mismatch can occur because a table was dropped in between creating the txn
- // and executing it.
- // It could also race with the table stopping/starting being managed by Accord.
- // The caller can retry if the table indeed exists and is managed by Accord.
- Set<TableId> txnDroppedTables = txnDroppedTables(keysOrRanges);
- Tracing.trace("Accord returned topology mismatch: " + coordinationFailed.getMessage());
- logger.debug("Accord returned topology mismatch", coordinationFailed);
- bookkeeping.markTopologyMismatch();
- // Throw IRE in case the caller fails to check if the table still exists
- if (!txnDroppedTables.isEmpty())
- {
- Tracing.trace("Accord txn uses dropped tables {}", txnDroppedTables);
- logger.debug("Accord txn uses dropped tables {}", txnDroppedTables);
- throw new InvalidRequestException("Accord transaction uses dropped tables");
- }
- trySuccess((V) RetryWithNewProtocolResult.instance);
- return false;
- }
else if (fail instanceof TimeoutException)
{
report = bookkeeping.newTimeout(txnId, keysOrRanges);
@@ -184,6 +164,26 @@ public class AccordResult<V> extends AsyncFuture<V> implements BiConsumer<V, Thr
if (txnId != null && txnId.isSyncPoint())
AccordAgent.onFailedBarrier(txnId, fail);
}
+ else if (isTxnRequest && fail instanceof TopologyMismatch)
+ {
+ // Excluding bugs topology mismatch can occur because a table was dropped in between creating the txn
+ // and executing it.
+ // It could also race with the table stopping/starting being managed by Accord.
+ // The caller can retry if the table indeed exists and is managed by Accord.
+ Set<TableId> txnDroppedTables = txnDroppedTables(keysOrRanges);
+ Tracing.trace("Accord returned topology mismatch: " + coordinationFailed.getMessage());
+ logger.debug("Accord returned topology mismatch", coordinationFailed);
+ bookkeeping.markTopologyMismatch();
+ // Throw IRE in case the caller fails to check if the table still exists
+ if (!txnDroppedTables.isEmpty())
+ {
+ Tracing.trace("Accord txn uses dropped tables {}", txnDroppedTables);
+ logger.debug("Accord txn uses dropped tables {}", txnDroppedTables);
+ throw new InvalidRequestException("Accord transaction uses dropped tables");
+ }
+ trySuccess((V) RetryWithNewProtocolResult.instance);
+ return false;
+ }
else if (fail instanceof RequestTimeoutException || fail instanceof TimeoutException)
{
report = bookkeeping.newTimeout(txnId, keysOrRanges);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index fa26ea24b6..cff5f26589 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -70,7 +70,6 @@ import accord.local.durability.DurabilityService;
import accord.local.durability.ShardDurability;
import accord.messages.Reply;
import accord.messages.Request;
-import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Seekable;
@@ -149,6 +148,7 @@ import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED;
import static org.apache.cassandra.journal.Params.ReplayMode.RESET;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadBookkeeping;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteBookkeeping;
+import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord;
import static org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.getTableMetadata;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@@ -233,7 +233,7 @@ public class AccordService implements IAccordService, Shutdownable
private final AccordResponseVerbHandler<? extends Reply> responseHandler;
@GuardedBy("this")
- private State state = State.INIT;
+ private volatile State state = State.INIT;
private static final IAccordService NOOP_SERVICE = new NoOpAccordService();
@@ -254,11 +254,26 @@ public class AccordService implements IAccordService, Shutdownable
unsafeInstance = instance = NOOP_SERVICE;
}
+ public static IAccordService tryGetUnsafe()
+ {
+ return unsafeInstance;
+ }
+
public static boolean isSetup()
{
return instance != null;
}
+ public static boolean isSetupOrStarting()
+ {
+ return unsafeInstance != null;
+ }
+
+ public static boolean isStarted()
+ {
+ return isSetup() && unsafeInstance instanceof AccordService && ((AccordService) unsafeInstance).state == State.STARTED;
+ }
+
public static IVerbHandler<Void> watermarkHandlerOrNoop()
{
if (!isSetup()) return ignore -> {};
@@ -268,38 +283,47 @@ public class AccordService implements IAccordService, Shutdownable
public static IVerbHandler<? extends Request> requestHandlerOrNoop()
{
- if (!isSetup()) return ignore -> {};
- return instance().requestHandler();
+ if (instance == null) return ignore -> {};
+ return instance.requestHandler();
}
public static IVerbHandler<? extends Reply> responseHandlerOrNoop()
{
- if (!isSetup()) return ignore -> {};
- return instance().responseHandler();
+ if (unsafeInstance == null) return ignore -> {};
+ return unsafeInstance.responseHandler();
}
@VisibleForTesting
- public synchronized static AccordService startup(NodeId tcmId)
+ public synchronized static void localStartup(NodeId tcmId)
{
+ Invariants.require(instance == null);
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
{
unsafeInstance = instance = NOOP_SERVICE;
- return null;
}
+ else
+ {
+ AccordService as = new AccordService(tcmIdToAccord(tcmId));
+ unsafeInstance = as;
+ as.localStartup();
+ }
+ }
- if (instance != null)
- return (AccordService) instance;
+ public synchronized static AccordService distributedStartup()
+ {
+ if (unsafeInstance == NOOP_SERVICE)
+ return null;
+
+ AccordService as = (AccordService) unsafeInstance;
+ if (as.state != State.INIT)
+ return as;
- AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
- unsafeInstance = as;
- as.startup();
+ as.distributedStartupInternal();
instance = as;
-
+
AccordReplicaMetrics.touch();
AccordSystemMetrics.touch();
AccordViolationHandler.setup();
-
- WatermarkCollector.fetchAndReportWatermarksAsync(as.topology());
return as;
}
@@ -323,12 +347,6 @@ public class AccordService implements IAccordService, Shutdownable
return true;
}
- @Override
- public boolean shouldAcceptMessages()
- {
- return state == State.STARTED && journal.started();
- }
-
public static IAccordService instance()
{
IAccordService i = instance;
@@ -365,7 +383,7 @@ public class AccordService implements IAccordService, Shutdownable
this.endpointMapper = new EndpointMapping.Updateable();
this.topologyService = new AccordTopologyService(localId, endpointMapper);
this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, endpointMapper);
- this.messageSink = new AccordMessageSink(agent, endpointMapper, callbacks);
+ this.messageSink = new AccordMessageSink(endpointMapper, callbacks);
this.node = new Node(localId,
messageSink,
topologyService,
@@ -391,7 +409,7 @@ public class AccordService implements IAccordService, Shutdownable
}
@Override
- public synchronized void startup()
+ public synchronized void localStartup()
{
if (state != State.INIT)
return;
@@ -428,8 +446,13 @@ public class AccordService implements IAccordService, Shutdownable
{
node.unsafeSetReplaying(false);
}
+ }
+
+ private void distributedStartupInternal()
+ {
+ finishTopologyInitialization();
+ WatermarkCollector.fetchAndReportWatermarksAsync(topology());
- finishInitialization();
catchup();
fastPathCoordinator.start();
@@ -439,9 +462,12 @@ public class AccordService implements IAccordService, Shutdownable
Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
- // Only enable durability scheduling _after_ we have fully replayed journal
+ // Only enable durability scheduling and progress logs _after_ we have fully replayed journal
node.durability().start();
+ // we set ourselves to STARTED before starting progress logs as this is the condition we use to decide if we
+ // start the progress log on command store initialisation (so creates a synchronisation point)
state = State.STARTED;
+ node.commandStores().forAll("", safeStore -> safeStore.progressLog().start());
}
void catchup()
@@ -477,13 +503,18 @@ public class AccordService implements IAccordService, Shutdownable
Throwable failed = f.cause();
if (failed != null)
- throw new RuntimeException("Could not catch up with peers", failed);
+ {
+ if (spec.catchup_on_start_exit_on_failure)
+ throw new RuntimeException("Could not catch up with peers", failed);
+
+ logger.error("Could not catch up with peers; continuing to startup");
+ break;
+ }
long end = nanoTime();
double seconds = NANOSECONDS.toMillis(end - start)/1000.0;
logger.info("Finished catching up with all quorums. {}s elapsed.", String.format("%.2f", seconds));
- // TODO (expected): make configurable
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
break;
@@ -511,9 +542,9 @@ public class AccordService implements IAccordService, Shutdownable
/**
* Startup is broken up in two phases: local and distributed startup. During local startup, we replay up to
* the latest epoch known to the node prior to restart. After that, we replay journal itself, and only after
- * that we finish initializaiton and replay the rest of epochs.
+ * that we finish initialization and replay the rest of epochs.
*/
- void finishInitialization()
+ void finishTopologyInitialization()
{
endpointMapper.updateMapping(ClusterMetadata.current());
TopologyManager topology = node.topology();
@@ -686,10 +717,9 @@ public class AccordService implements IAccordService, Shutdownable
private AsyncChain<Void> syncInternal(Timestamp minBound, Keys keys, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote)
{
TxnId txnId = node.nextTxnId(minBound, keys, Write);
- FullRoute<?> route = node.computeRoute(txnId, keys);
return node.withEpochAtLeast(txnId.epoch(), null, () -> {
Txn txn = new Txn.InMemory(Write, keys, TxnRead.createNoOpRead(keys), TxnQuery.UNSAFE_EMPTY, TxnUpdate.empty(), new TableMetadatasAndKeys(TableMetadatas.none(), keys));
- return CoordinateTransaction.coordinate(node, route, txnId, txn)
+ return CoordinateTransaction.coordinate(node, txnId, txn)
.mapToNull();
});
}
@@ -782,7 +812,7 @@ public class AccordService implements IAccordService, Shutdownable
return keys;
TableId tableId = tableId(keys, r -> ((AccordRoutableKey)r).table());
- return sliceToAccord(tableId, keys, Keys::slice);
+ return sliceToAccord(tableId, keys, Keys::overlapping);
}
public static Ranges intersecting(Ranges ranges)
@@ -791,7 +821,7 @@ public class AccordService implements IAccordService, Shutdownable
return ranges;
TableId tableId = tableId(ranges, r -> ((TokenRange)r).table());
- return sliceToAccord(tableId, ranges, Ranges::slice);
+ return sliceToAccord(tableId, ranges, Ranges::overlapping);
}
private static <C extends Seekables<?, ?>> C sliceToAccord(TableId tableId, C collection, BiFunction<C, Ranges, C> slice)
@@ -974,7 +1004,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public void shutdownAndWait(long timeout, TimeUnit unit)
{
- if (!ExecutorUtils.shutdownSequentiallyAndWait(shutdownableSubsystems(), timeout, unit))
+ if (!ExecutorUtils.shutdownThenWait(shutdownableSubsystems(), timeout, unit))
logger.error("One or more subsystems did not shut down cleanly.");
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java
index 2234ca00f4..79c1170827 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTask.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java
@@ -728,12 +728,12 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
public void fail(Throwable throwable)
{
- commandStore.agent().onUncaughtException(throwable);
+ commandStore.agent().onException(throwable);
if (state.isComplete())
return;
if (commandStore.hasSafeStore())
- commandStore.agent().onUncaughtException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
+ commandStore.agent().onException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
state(FAILING);
if (callback != null)
@@ -747,12 +747,12 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
{
if (newFailure)
{
- commandStore.agent().onUncaughtException(throwable);
+ commandStore.agent().onException(throwable);
if (state.isComplete())
return;
if (commandStore.hasSafeStore())
- commandStore.agent().onUncaughtException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
+ commandStore.agent().onException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
}
state(FAILED);
@@ -860,7 +860,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
catch (Throwable t)
{
releaseResourcesSlow(caches, t);
- commandStore.agent().onUncaughtException(t);
+ commandStore.agent().onException(t);
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTracing.java b/src/java/org/apache/cassandra/service/accord/AccordTracing.java
index ff8bddfe8f..c1a69d568d 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTracing.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTracing.java
@@ -760,6 +760,9 @@ public class AccordTracing extends AccordCoordinatorMetrics.Listener
public Tracing trace(TxnId txnId, @Nullable Participants<?> participants, CoordinationKind kind)
{
+ if (kind == CoordinationKind.FetchDurableBefore)
+ return (cs, msg) -> logger.info("Catchup/FetchDurableBefore: {}", msg);
+
if (!txnIdMap.containsKey(txnId) && null == maybeTrace(txnId, participants, kind, NewOrFailure.NEW, traceNewPatterns))
return null;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
index a850a537fd..4833c5beec 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
@@ -46,12 +46,6 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
@Override
public void doVerb(Message<T> message) throws IOException
{
- if (!AccordService.instance().shouldAcceptMessages())
- {
- dropping.debug(message.verb(), message.from());
- return;
- }
-
logger.trace("Receiving {} from {}", message.payload, message.from());
T request = message.payload;
diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java
index 8c74725ab6..e500cd94a5 100644
--- a/src/java/org/apache/cassandra/service/accord/IAccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java
@@ -108,6 +108,7 @@ public interface IAccordService
}
boolean isEnabled();
+
long currentEpoch();
void setCacheSize(long kb);
@@ -115,7 +116,7 @@ public interface IAccordService
TopologyManager topology();
- void startup();
+ void localStartup();
Future<Void> flushCaches();
void markShuttingDown();
@@ -186,8 +187,6 @@ public interface IAccordService
Params journalConfiguration();
- boolean shouldAcceptMessages();
-
Node node();
/**
@@ -274,7 +273,7 @@ public interface IAccordService
}
@Override
- public void startup()
+ public void localStartup()
{
try
{
@@ -369,12 +368,6 @@ public interface IAccordService
throw new UnsupportedOperationException("Cannot return configuration when accord.enabled = false in cassandra.yaml");
}
- @Override
- public boolean shouldAcceptMessages()
- {
- return true;
- }
-
@Override
public Node node()
{
@@ -490,9 +483,9 @@ public interface IAccordService
}
@Override
- public void startup()
+ public void localStartup()
{
- delegate.startup();
+ delegate.localStartup();
}
@Override
@@ -573,12 +566,6 @@ public interface IAccordService
return delegate.journalConfiguration();
}
- @Override
- public boolean shouldAcceptMessages()
- {
- return delegate.shouldAcceptMessages();
- }
-
@Override
public Node node()
{
diff --git a/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java b/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java
index 0f1e82a0d3..cc214dd380 100644
--- a/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java
+++ b/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java
@@ -36,7 +36,7 @@ public class ImmediateAsyncExecutor implements AbstractAsyncExecutor
}
catch (Throwable t)
{
- AccordAgent.handleUncaughtException(t);
+ AccordAgent.handleException(t);
}
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java
index a4ed02eaa1..12229c99dc 100644
--- a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java
+++ b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java
@@ -139,7 +139,6 @@ public class WatermarkCollector implements TopologyListener
return;
Snapshot snapshot = m.payload;
- long minEpoch = topologyManager.minEpoch();
forEachEpoch(topologyManager::onEpochClosed, snapshot.closed);
forEachEpoch(topologyManager::onEpochRetired, snapshot.retired);
for (Map.Entry<Long, Long> e : snapshot.synced.entrySet())
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
index e310353729..636535e576 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord.api;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
@@ -36,6 +37,7 @@ import accord.api.ProgressLog.BlockedUntil;
import accord.api.RoutingKey;
import accord.api.Tracing;
import accord.coordinate.Coordination;
+import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.Node;
import accord.local.SafeCommand;
@@ -193,24 +195,23 @@ public class AccordAgent implements Agent, OwnershipEventListener
logger.error("This replica has become stale for {} as of {}", ranges, staleSince);
}
- @Override
- public void onUncaughtException(Throwable t)
+ public static void handleException(Throwable t)
{
- handleUncaughtException(t);
+ if (t instanceof RequestTimeoutException || t instanceof CancellationException || t instanceof TimeoutException || t instanceof Timeout)
+ return;
+ JVMStabilityInspector.uncaughtException(Thread.currentThread(), t);
}
- public static void handleUncaughtException(Throwable t)
+ @Override
+ public void onException(Throwable t)
{
- if (t instanceof RequestTimeoutException || t instanceof CancellationException)
- return;
- JVMStabilityInspector.uncaughtException(Thread.currentThread(), t);
+ handleException(t);
}
@Override
- public void onCaughtException(Throwable t, String context)
+ public void onException(Throwable t, String context)
{
- logger.warn(context, t);
- JVMStabilityInspector.inspectThrowable(t);
+ handleException(t);
}
@Override
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 d69abfd96e..2c5f405205 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
@@ -106,8 +106,18 @@ 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, ballot, executeAt, deps, callback, executor, consistencyLevel, endpointMapper)
- .start();
+ AccordInteropExecution execution;
+ try
+ {
+ execution = new AccordInteropExecution(node, txnId, txn, updateKind, route, ballot, executeAt, deps, callback, executor, consistencyLevel, endpointMapper);
+ }
+ catch (Throwable t)
+ {
+ callback.accept(null, t);
+ return true;
+ }
+
+ execution.start();
return true;
}
@@ -119,9 +129,19 @@ public class AccordInteropAdapter extends TxnAdapter
if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty())
return false;
- Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt);
- new AccordInteropPersist(node, executor, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, CoordinationFlags.none(), informDurableOnDone, Minimal, callback)
- .start();
+ AccordInteropPersist persist;
+ try
+ {
+ Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt);
+ persist = new AccordInteropPersist(node, executor, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, CoordinationFlags.none(), informDurableOnDone, Minimal, callback);
+ }
+ catch (Throwable t)
+ {
+ callback.accept(null, t);
+ return true;
+ }
+
+ persist.start();
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 cd83f3ee7d..9fef927579 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
@@ -178,7 +178,7 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
else if (failure != null)
{
node.reply(replyTo, replyContext, null, failure);
- node.agent().onUncaughtException(failure);
+ node.agent().onException(failure);
fail();
}
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 eaae5c350c..cbc6182674 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
@@ -50,6 +50,7 @@ import accord.topology.ActiveEpochs;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
+import accord.topology.TopologyException;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncChain;
@@ -139,7 +140,7 @@ public class AccordInteropExecution implements ReadCoordinator
private volatile long uniqueHlc;
public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute<?> route, Ballot ballot, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback,
- SequentialAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper)
+ SequentialAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) throws TopologyException
{
requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR);
this.node = node;
diff --git a/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java b/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java
index c2e9bce8aa..d9a905bb43 100644
--- a/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java
+++ b/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java
@@ -53,7 +53,6 @@ import org.apache.cassandra.utils.concurrent.Future;
import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
import static accord.local.durability.DurabilityService.SyncRemote.All;
-import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static accord.primitives.Timestamp.mergeMax;
import static accord.primitives.Timestamp.minForEpoch;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -79,15 +78,15 @@ public class AccordRepair
private volatile Throwable shouldAbort = null;
private volatile Thread waiting;
- public AccordRepair(SharedContext ctx, ColumnFamilyStore cfs, TimeUUID repairId, String keyspace, Collection<Range<Token>> ranges, boolean requireAllEndpoints, List<InetAddressAndPort> endpoints)
+ public AccordRepair(SharedContext ctx, ColumnFamilyStore cfs, TimeUUID repairId, String keyspace, Collection<Range<Token>> ranges, SyncRemote syncRemote, List<InetAddressAndPort> endpoints)
{
this.ctx = ctx;
this.cfs = cfs;
this.repairId = repairId;
+ this.syncRemote = syncRemote;
// TODO (desired): support unsafe configuration where we permit less than a quorum,
// but this is challenging to do safely as repair has no concept of participants from earlier epochs
- this.syncRemote = requireAllEndpoints ? All : Quorum;
- if (endpoints != null)
+ if (syncRemote != All && endpoints != null)
{
including = new ArrayList<>(endpoints.size());
AccordEndpointMapper mapper = AccordService.instance().endpointMapper();
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AbstractSortedCollector.java b/src/java/org/apache/cassandra/service/accord/serializers/AbstractSortedCollector.java
index 80672054ed..2183905fe7 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/AbstractSortedCollector.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/AbstractSortedCollector.java
@@ -97,34 +97,38 @@ public abstract class AbstractSortedCollector<T, C> extends AbstractList<T>
return add;
}
- public void clear()
- {
- if (count > 1)
- cachedAny().forceDiscard((Object[])buffer, count);
- buffer = null;
- count = 0;
- }
-
public C build()
{
+ C result;
if (count == 0)
{
- return empty();
+ result = empty();
}
else if (count == 1)
{
- return of((T)buffer);
+ result = of((T)buffer);
}
else if (count < BTREE_THRESHOLD)
{
- C result = copy((Object[])buffer, count);
+ result = copy((Object[])buffer, count);
cachedAny().forceDiscard((Object[])buffer, count);
- return result;
}
else
{
- return copyBtree((Object[])buffer, count);
+ result = copyBtree((Object[])buffer, count);
}
+ buffer = null;
+ count = 0;
+ return result;
+ }
+
+
+ public void clear()
+ {
+ if (count > 1 && count < BTREE_THRESHOLD)
+ cachedAny().forceDiscard((Object[])buffer, count);
+ buffer = null;
+ count = 0;
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
index e9b42c338f..5163ab5418 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
@@ -168,7 +168,7 @@ public class CommandSerializers
if (executeAt.equals(Timestamp.NONE))
return size + TypeSizes.sizeofUnsignedVInt(0L);
- size += TypeSizes.sizeofUnsignedVInt(executeAt.epoch() - txnId.epoch());
+ size += TypeSizes.sizeofUnsignedVInt(1 + executeAt.epoch() - txnId.epoch());
}
size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc() - txnId.hlc());
size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id);
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 2affc5d778..d4b549a61d 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
@@ -67,6 +67,7 @@ import org.apache.cassandra.utils.CollectionSerializers;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.Pair;
+import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.ArrayBuffers.cachedInts;
import static accord.utils.Invariants.requireArgument;
import static accord.utils.SortedArrays.Search.CEIL;
@@ -637,13 +638,13 @@ public class TxnUpdate extends AccordUpdate
@Override
public TxnUpdate slice(Ranges ranges)
{
- return getTxnUpdate(keys -> keys.slice(ranges));
+ return getTxnUpdate(keys -> keys.slice(ranges, Minimal));
}
@Override
public TxnUpdate intersecting(Participants<?> participants)
{
- return getTxnUpdate(keys -> keys.intersecting(participants));
+ return getTxnUpdate(keys -> keys.intersecting(participants, Minimal));
}
@VisibleForTesting
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java
index 26af2aaaaf..e00da6690a 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java
@@ -68,7 +68,6 @@ import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.serializers.TableMetadatas;
import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
import org.apache.cassandra.service.accord.serializers.Version;
-import org.apache.cassandra.utils.BooleanSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.SimpleBitSetSerializers;
@@ -545,7 +544,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
public void skip(Seekables keys, DataInputPlus in, Version version) throws IOException
{
TableMetadatas tables = TableMetadatas.deserializeSelf(in);
- BooleanSerializer.serializer.deserialize(in);
+ SimpleBitSetSerializers.any.deserialize(in);
skipArray(new TableMetadatasAndKeys(tables, keys), in, version, Update.serializer);
}
diff --git a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java
index 07b64aa57f..79a7f67ed5 100644
--- a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java
+++ b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java
@@ -380,7 +380,7 @@ public abstract class ConsensusTableMigration
boolean optimiseStreams = false;
boolean ignoreUnreplicatedKeyspaces = true;
boolean dontPurgeTombstones = false;
- RepairOption repairOption = new RepairOption(RepairParallelism.PARALLEL, primaryRange, incremental, trace, numJobThreads, intersectingRanges, pullRepair, forceRepair, PreviewKind.NONE, optimiseStreams, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord);
+ RepairOption repairOption = new RepairOption(RepairParallelism.PARALLEL, primaryRange, incremental, trace, numJobThreads, intersectingRanges, pullRepair, forceRepair, PreviewKind.NONE, optimiseStreams, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord, false);
tables.forEach(table -> repairOption.getColumnFamilies().add(table.tableName));
return repairOption;
}
diff --git a/src/java/org/apache/cassandra/tools/nodetool/Repair.java b/src/java/org/apache/cassandra/tools/nodetool/Repair.java
index 8836f00582..1d0aa8e95f 100644
--- a/src/java/org/apache/cassandra/tools/nodetool/Repair.java
+++ b/src/java/org/apache/cassandra/tools/nodetool/Repair.java
@@ -117,6 +117,9 @@ public class Repair extends AbstractCommand
@Option(paramLabel = "accord-only", names = { "-accord-only", "--accord-only" }, description = "If the --accord-only flag is included, no table data is repaired, only accord operations..")
private boolean accordOnly = false;
+ @Option(paramLabel = "permit-no-quorum", names = { "-permit-no-quorum", "--permit-no-quorum" }, description = "If the --permit-no-quorum flag is included, accord repair is permitted to proceed without a quorum, including only the specified nodes")
+ private boolean permitNoQuorum = false;
+
@Option(paramLabel = "skip-accord", names = { "-skip-accord", "--skip-accord" }, description = "If the --skip-accord flag is included, the Accord repair step is skipped. Accord repair is also skipped for preview repairs.")
private boolean skipAccord = false;
@@ -202,6 +205,7 @@ public class Repair extends AbstractCommand
checkArgument(!(skipAccord && accordOnly), "Can't specify both skip-accord and accord-only");
boolean repairAccord = !skipAccord && !paxosOnly && getPreviewKind() == PreviewKind.NONE;
options.put(RepairOption.REPAIR_ACCORD_KEY, Boolean.toString(repairAccord));
+ options.put(RepairOption.PERMIT_NO_QUORUM_KEY, Boolean.toString(permitNoQuorum));
boolean repairData = false;
if (getPreviewKind() == PreviewKind.NONE)
{
diff --git a/src/java/org/apache/cassandra/utils/ExecutorUtils.java b/src/java/org/apache/cassandra/utils/ExecutorUtils.java
index a28225456c..30a67fbe3c 100644
--- a/src/java/org/apache/cassandra/utils/ExecutorUtils.java
+++ b/src/java/org/apache/cassandra/utils/ExecutorUtils.java
@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@@ -103,15 +104,51 @@ public class ExecutorUtils
else
throw new IllegalArgumentException(executor.toString());
}
- catch (Throwable t)
+ catch (InterruptedException t)
{
- throw new IllegalStateException("Caught interrupt while shutting down " + executor, t);
+ throw new UncheckedInterruptedException(t);
}
}
return shutdown;
}
+ public static boolean shutdownThenWait(Iterable<?> executors, long timeout, TimeUnit unit)
+ {
+ long deadline = nanoTime() + unit.toNanos(timeout);
+
+ for (Object executor : executors)
+ {
+ if (executor instanceof ExecutorService) ((ExecutorService) executor).shutdown();
+ else if (executor instanceof Shutdownable) ((Shutdownable) executor).shutdown();
+ else throw new IllegalArgumentException(executor.toString());
+ }
+
+ for (Object executor : executors)
+ {
+ try
+ {
+ if (executor instanceof ExecutorService)
+ {
+ if (!((ExecutorService) executor).awaitTermination(Math.max(0, deadline - nanoTime()), NANOSECONDS))
+ return false;
+ }
+ else if (executor instanceof Shutdownable)
+ {
+ if (!((Shutdownable) executor).awaitTermination(Math.max(0, deadline - nanoTime()), NANOSECONDS))
+ return false;
+ }
+ else throw new IllegalArgumentException(executor.toString());
+ }
+ catch (InterruptedException t)
+ {
+ throw new IllegalStateException("Caught interrupt while shutting down " + executor, t);
+ }
+ }
+
+ return true;
+ }
+
public static void shutdown(ExecutorService ... executors)
{
shutdown(Arrays.asList(executors));
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index 533a7b49d5..8ecdb5d267 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -45,6 +45,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -197,7 +198,12 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
CassandraRelevantProperties.TEST_FLUSH_LOCAL_SCHEMA_CHANGES.reset();
CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
CassandraRelevantProperties.IO_NETTY_TRANSPORT_NONATIVE.setBoolean(false);
- withInstanceInitializer((classLoader, threadGroup, i, i1) -> {
+ withInstanceInitializer(defaultInitializer());
+ }
+
+ private IInstanceInitializer defaultInitializer()
+ {
+ return (classLoader, threadGroup, i, i1) -> {
try
{
Class<?> ef = classLoader.loadClass(ExecutorFactory.class.getName());
@@ -215,10 +221,21 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
else
logger.info("Unable to set ExecutorFactory for instance {}", i, e);
}
- catch (NoSuchMethodException | InvocationTargetException | InstantiationException | IllegalAccessException e)
+ catch (NoSuchMethodException | InvocationTargetException | InstantiationException |
+ IllegalAccessException e)
{
throw new RuntimeException(e);
}
+ };
+ }
+
+ @Override
+ public B withInstanceInitializer(BiConsumer<ClassLoader, Integer> instanceInitializer)
+ {
+ IInstanceInitializer wrap = defaultInitializer();
+ return withInstanceInitializer((classLoader, threadGroup, num, v) -> {
+ wrap.initialise(classLoader, threadGroup, num, v);
+ instanceInitializer.accept(classLoader, num);
});
}
@@ -573,6 +590,10 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
protected AbstractCluster(AbstractBuilder<I, ? extends ICluster<I>, ?> builder)
{
+ // start the JNA cleaner on the system class loader to avoid pinning an instance
+ // (we do it here because startup() isn't always called)
+ com.sun.jna.internal.Cleaner.getCleaner();
+
this.root = builder.getRootPath();
this.sharedClassLoader = builder.getSharedClassLoader();
this.sharedClassPredicate = builder.getSharedClasses();
@@ -1068,8 +1089,6 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
public void startup()
{
- // start the JNA cleaner on the system class loader to avoid pinning an instance
- com.sun.jna.internal.Cleaner.getCleaner();
try (AllMembersAliveMonitor monitor = new AllMembersAliveMonitor())
{
monitor.startPolling();
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 6147a3fdd3..f0a35d6d83 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -51,6 +51,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.utils.Invariants;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.cassandra.Util;
import org.apache.cassandra.audit.AuditLogManager;
@@ -64,6 +65,7 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.SharedExecutorPool;
import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
@@ -822,6 +824,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
try
{
CommitLog.instance.recoverSegmentsOnDisk();
+ NodeId self = ClusterMetadata.current().myNodeId();
+ if (self != null)
+ AccordService.localStartup(self);
}
catch (IOException e)
{
@@ -868,7 +873,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
ClusterMetadataService.instance().processor().fetchLogAndWait();
NodeId self = Register.maybeRegister();
RegistrationStatus.instance.onRegistration();
- AccordService.startup(self);
+ if (!AccordService.isSetupOrStarting())
+ AccordService.localStartup(self);
+ AccordService.distributedStartup();
+
boolean joinRing = config.get(Constants.KEY_DTEST_JOIN_RING) == null || (boolean) config.get(Constants.KEY_DTEST_JOIN_RING);
if (ClusterMetadata.current().directory.peerState(self) != NodeState.JOINED && joinRing)
{
@@ -1028,8 +1036,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
);
error = parallelRun(error, executor, () -> {
- if (!AccordService.isSetup()) return;
- AccordService.instance().shutdownAndWait(1l, MINUTES);
+ if (AccordService.isSetupOrStarting())
+ AccordService.unsafeInstance().shutdownAndWait(1L, MINUTES);
});
// CommitLog must shut down after Stage, or threads from the latter may attempt to use the former.
@@ -1064,6 +1072,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
try
{
future.get();
+ ThreadGroup group = Thread.currentThread().getThreadGroup();
+ int active = group.activeCount();
+ Invariants.expect(group.getParent().activeCount() <= active
+ || CassandraRelevantProperties.DTEST_IGNORE_SHUTDOWN_THREADCOUNT.getBoolean());
return null;
}
finally
diff --git a/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java b/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
index ddee1fef63..fa25d691e4 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
+import org.junit.BeforeClass;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
@@ -53,6 +54,12 @@ import static org.junit.Assert.fail;
public class DecommissionTest extends TestBaseImpl
{
+ @BeforeClass
+ public static void before()
+ {
+ CassandraRelevantProperties.DTEST_IGNORE_SHUTDOWN_THREADCOUNT.setBoolean(true);
+ }
+
@Test
public void testDecommission() throws Throwable
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java
index 2279026306..2577d02a7a 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java
@@ -48,6 +48,7 @@ import org.slf4j.LoggerFactory;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
+import accord.topology.TopologyException;
import org.apache.cassandra.config.Config.PaxosVariant;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ast.Symbol;
@@ -709,7 +710,9 @@ public abstract class AccordCQLTestBase extends AccordTestBase
Unseekables<?> routables = AccordTestUtils.createTxn(sb.toString()).keys().toParticipants();
long epoch = AccordService.instance().topology().epoch();
- Topologies topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch);
+ Topologies topology;
+ try { topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch); }
+ catch (TopologyException e) { throw new RuntimeException(e); }
// we don't detect out-of-bounds read/write yet, so use this to validate we reach different shards
Assertions.assertThat(topology.totalShards()).isEqualTo(2);
});
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCommandStoreTryExecuteListeningTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCommandStoreTryExecuteListeningTest.java
index 90f22d5abc..ea7e0d871d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCommandStoreTryExecuteListeningTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCommandStoreTryExecuteListeningTest.java
@@ -38,6 +38,7 @@ import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Txn;
import accord.primitives.TxnId;
+import accord.topology.TopologyException;
import accord.utils.ImmutableBitSet;
import accord.utils.LargeBitSet;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -127,7 +128,9 @@ public class AccordCommandStoreTryExecuteListeningTest extends TestBaseImpl
PartitionKey key = pk(1, "ks", "tbl");
Txn txn = node.agent().emptySystemTxn(Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range);
TxnId txnId = node.nextTxnId(txn);
- FullRoute<?> route = node.computeRoute(txnId, Ranges.of(key.asRange()));
+ FullRoute<?> route;
+ try { route = node.computeRoute(txnId, Ranges.of(key.asRange())); }
+ catch (TopologyException e) { throw new RuntimeException(e); }
AccordCommandStore commandStore = (AccordCommandStore) node.commandStores().unsafeForKey(key.toUnseekable());
int[] rangesToTxnIds = new int[dependencies.length + 1];
rangesToTxnIds[0] = rangesToTxnIds.length;
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java
index ff47950f7a..9efb7bb035 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java
@@ -194,7 +194,6 @@ public class AccordMetricsTest extends AccordTestBase
assertClientMetrics(0, "AccordRead", 1, 0);
assertClientMetrics(1, "AccordRead", 0, 0);
- // TODO (required): reenable or remove metric
assertCoordinatorMetrics(0, "ro", 0, 0, 1, 0, 0);
assertCoordinatorMetrics(1, "ro" , 0, 0, 0, 0, 1);
assertReplicaMetrics(0, "ro", 0, 0, 0);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java
index 1a8a8a7223..8443526e9f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java
@@ -110,7 +110,7 @@ public class AccordSimpleFastPathTest extends TestBaseImpl
long epoch = cm.epoch.getEpoch();
TopologyManager tm = AccordService.instance().topology();
- Topology topology = tm.active().globalForEpoch(epoch);
+ Topology topology = tm.active().getKnown(epoch).global();
Assert.assertFalse(topology.shards().isEmpty());
topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath)));
return cm.epoch.getEpoch();
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
index efabf33022..ba11a39a69 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
@@ -43,6 +43,7 @@ import accord.primitives.SyncPoint;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Writes;
+import accord.topology.TopologyException;
import accord.utils.ImmutableBitSet;
import accord.utils.LargeBitSet;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -114,7 +115,9 @@ public class AccordJournalReplayTest extends TestBaseImpl
Txn syncPointTxn = node.agent().emptySystemTxn(Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range);
TxnId syncPointId = node.nextTxnId(syncPointTxn);
TxnId txnId = node.nextTxnId(txn);
- FullRoute<?> route = node.computeRoute(txnId, txn.keys());
+ FullRoute<?> route;
+ try { route = node.computeRoute(txnId, txn.keys()); }
+ catch (TopologyException e) { throw new RuntimeException(e); }
AccordCommandStore commandStore = (AccordCommandStore) node.commandStores().unsafeForKey(key.toUnseekable());
Deps deps = new Deps(KeyDeps.NONE, RangeDeps.SerializerSupport.create(new accord.primitives.Range[] { key.asRange() }, new TxnId[] { syncPointId }, new int[] { 2, 0 }, new int[] { 2, 0 }));
Command.WaitingOn waitingOn; {
diff --git a/test/resources/nodetool/help/repair b/test/resources/nodetool/help/repair
index 7710ab4091..3890077a42 100644
--- a/test/resources/nodetool/help/repair
+++ b/test/resources/nodetool/help/repair
@@ -16,7 +16,8 @@ SYNOPSIS
[(-iuk | --ignore-unreplicated-keyspaces)]
[(-j <job_threads> | --job-threads <job_threads>)]
[(-local | --in-local-dc)] [(-os | --optimise-streams)]
- [(-paxos-only | --paxos-only)] [(-pl | --pull)]
+ [(-paxos-only | --paxos-only)]
+ [(-permit-no-quorum | --permit-no-quorum)] [(-pl | --pull)]
[(-pr | --partitioner-range)] [(-prv | --preview)]
[(-seq | --sequential)] [(-skip-accord | --skip-accord)]
[(-skip-paxos | --skip-paxos)]
@@ -76,6 +77,11 @@ OPTIONS
If the --paxos-only flag is included, no table data is repaired,
only paxos operations..
+ -permit-no-quorum, --permit-no-quorum
+ If the --permit-no-quorum flag is included, accord repair is
+ permitted to proceed without a quorum, including only the specified
+ nodes
+
-pl, --pull
Use --pull to perform a one way repair where data is only streamed
from a remote node to this node.
diff --git a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceRepair.java b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceRepair.java
index 55ab4f30a8..e69923e86d 100644
--- a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceRepair.java
+++ b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceRepair.java
@@ -129,7 +129,7 @@ public class OnInstanceRepair extends ClusterAction
{
Collection<Range<Token>> ranges = rangesSupplier.call();
// no need to wait for completion, as we track all task submissions and message exchanges, and ensure they finish before continuing to next action
- StorageService.instance.repair(keyspaceName, new RepairOption(RepairParallelism.SEQUENTIAL, isPrimaryRangeOnly, repairType.incremental, false, 1, ranges, false, force, PreviewKind.NONE, false, true, repairType.repairData, repairType.repairPaxos, false, repairType.repairAccord), singletonList((tag, event) -> {
+ StorageService.instance.repair(keyspaceName, new RepairOption(RepairParallelism.SEQUENTIAL, isPrimaryRangeOnly, repairType.incremental, false, 1, ranges, false, force, PreviewKind.NONE, false, true, repairType.repairData, repairType.repairPaxos, false, repairType.repairAccord, false), singletonList((tag, event) -> {
if (event.getType() == ProgressEventType.COMPLETE)
listener.run();
}));
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
index 39b693dad4..e483ff49ca 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
@@ -279,10 +279,14 @@ public class CompactionAccordIteratorsTest
private static void flush(AccordCommandStore commandStore)
{
+ long cacheSize;
try (AccordExecutor.ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
- long cacheSize = cache.global.capacity();
+ cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
+ }
+ try (AccordExecutor.ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
+ {
cache.global.setCapacity(cacheSize);
}
commandsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
@@ -313,7 +317,7 @@ public class CompactionAccordIteratorsTest
Txn txn = txnId.kind().isWrite() ? writeTxn : readTxn;
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
- Route<?> partialRoute = route.slice(commandStore.unsafeGetRangesForEpoch().currentRanges());
+ Route<?> partialRoute = route.overlapping(commandStore.unsafeGetRangesForEpoch().currentRanges());
getBlocking(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, (a, b) -> {});
}));
diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
index a496ac77f0..260d468af3 100644
--- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
@@ -37,6 +37,7 @@ import org.junit.Test;
import accord.api.ProtocolModifiers;
import accord.messages.NoWaitRequest;
import accord.api.RoutingKey;
+import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.SaveStatus;
@@ -209,7 +210,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
CQLTester.setUpClass();
CassandraDaemon.getInstanceForTesting().setupVirtualKeyspaces();
- AccordService.startup(ClusterMetadata.current().myNodeId());
+ AccordService.localStartup(ClusterMetadata.current().myNodeId());
+ AccordService.distributedStartup();
requireNetwork();
}
@@ -235,8 +237,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
MessagingService.instance().outboundSink.add(filter);
try
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
execute(SET_TRACE, 1, "{WaitProgress}", id.toString());
@@ -276,8 +278,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
MessagingService.instance().outboundSink.add(filter);
try
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
execute(SET_TRACE_REMOTE, 1, "{WaitProgress}", nodeId, id.toString());
@@ -339,8 +341,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.toString(), "*", "{}", "*", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count + 1 ; ++i)
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
@@ -356,8 +358,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.asRange().toString(), "{KE}", "{}", "{PreAccept}", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count ; ++i)
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
assertRows(execute(QUERY_TRACE, id.toString()));
}
@@ -365,8 +367,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
List<TxnId> txnIds = new ArrayList<>();
for (int i = 0 ; i < count + 1 ; ++i)
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.EphemeralRead, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("SELECT * FROM %s.%s WHERE k = ? AND c = ?", KEYSPACE, tableName)), 0, i);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.EphemeralRead, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
@@ -379,8 +381,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
}
{
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, "" + txn.keys().get(0).toUnseekable(), "{KW}", "*", "{}", "leaky", 1, 1, "{}", 1);
AccordMsgFilter filter = new AccordMsgFilter();
@@ -461,7 +463,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
AccordService accord = accord();
int nodeId = accord.nodeId().id;
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(Keys.of(), Txn.Kind.Write, Routable.Domain.Key);
try
{
execute(ERASE_JOURNAL_REMOTE, nodeId, 1, id.toString());
@@ -479,8 +481,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
AccordService accord = accord();
int nodeId = accord.nodeId().id;
AccordMsgFilter filter = new AccordMsgFilter();
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
filter.dropVerbs = Set.of();
MessagingService.instance().outboundSink.add(filter);
@@ -562,7 +564,6 @@ public class AccordDebugKeyspaceTest extends CQLTester
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
" IF r IS NULL THEN\n " +
@@ -570,6 +571,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
" END IF\n" +
"COMMIT TRANSACTION", KEYSPACE, tableName, KEYSPACE, tableName);
Txn txn = createTxn(insertTxn, 0, 0, 0, 0, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
accord.node().coordinate(id, txn).beginAsResult();
@@ -603,16 +605,17 @@ public class AccordDebugKeyspaceTest extends CQLTester
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
- TxnId first = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
" IF r IS NULL THEN\n " +
" INSERT INTO %s.%s (k, c, v) VALUES (?, ?, ?);\n" +
" END IF\n" +
"COMMIT TRANSACTION", KEYSPACE, tableName, KEYSPACE, tableName);
+ Txn txn = createTxn(insertTxn, 0, 0, 0, 0, 0);
+ TxnId first = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(first);
- accord.node().coordinate(first, createTxn(insertTxn, 0, 0, 0, 0, 0)).beginAsResult();
+ accord.node().coordinate(first, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
@@ -627,10 +630,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
filter.reset();
- TxnId second = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
+ TxnId second = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.reset();
filter.appliesTo(second);
- accord.node().coordinate(second, createTxn(insertTxn, 0, 0, 0, 0, 0)).beginAsResult();
+ accord.node().coordinate(second, txn).beginAsResult();
filter.commit.awaitThrowUncheckedOnInterrupt();
@@ -695,9 +698,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
KEYSPACE,
tableName);
AccordService accord = accord();
- TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
+ Txn txn = createTxn(insertTxn, 0, 0, 0);
+ TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
- accord.node().coordinate(id, createTxn(insertTxn, 0, 0, 0)).beginAsResult();
+ accord.node().coordinate(id, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
String QUERY_JOURNAL = String.format("SELECT txn_id, save_status, command_store_id FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);
diff --git a/test/unit/org/apache/cassandra/db/virtual/LazyVirtualTableTest.java b/test/unit/org/apache/cassandra/db/virtual/LazyVirtualTableTest.java
new file mode 100644
index 0000000000..496659c728
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/virtual/LazyVirtualTableTest.java
@@ -0,0 +1,351 @@
+/*
+ * 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.virtual;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntUnaryOperator;
+import java.util.stream.IntStream;
+
+import org.junit.Test;
+
+import accord.utils.Invariants;
+import accord.utils.LargeBitSet;
+import accord.utils.RandomSource;
+import accord.utils.RandomTestRunner;
+import accord.utils.UnhandledEnum;
+import net.openhft.chronicle.core.util.IntBiPredicate;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.schema.TableMetadata;
+import org.assertj.core.api.Assertions;
+
+import static org.apache.cassandra.db.virtual.AbstractLazyVirtualTable.OnTimeout.FAIL;
+import static org.apache.cassandra.db.virtual.LazyVirtualTableTest.Cmp.CMPS;
+import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.ASC;
+import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.DESC;
+import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.UNSORTED;
+
+public class LazyVirtualTableTest extends CQLTester
+{
+ private static final String KEYSPACE = "system_accord_not_really";
+
+ @Test
+ public void test()
+ {
+ DatabaseDescriptor.setRpcTimeout(TimeUnit.MINUTES.toMillis(5L));
+ RandomTestRunner.test().withSeed(3606318398900145565L).check(this::testOne);
+ for (int i = 0 ; i < 10 ; ++i)
+ RandomTestRunner.test().check(this::testOne);
+ }
+
+ private void testOne(RandomSource rnd)
+ {
+ QueryProcessor.clearInternalStatementsCache();
+ QueryProcessor.clearPreparedStatementsCache();
+ int keyCount = rnd.nextInt(1, 5);
+ int valCount = rnd.nextInt(0, 3);
+ int rowCount = rnd.nextInt(0, 12) - 1;
+ rowCount = rowCount < 0 ? 0 : rnd.nextInt(1 << rowCount, 2 << rowCount);
+ int[] keyDomains = new int[keyCount];
+ Arrays.fill(keyDomains, 1);
+ int uniqueKeys = Math.max(1, rowCount * rnd.nextInt(2, 5));
+ for (int k = keyDomains.length - 1 ; k > 0 && uniqueKeys > 1 ; --k)
+ {
+ int maxScale = 31 - Integer.numberOfLeadingZeros(uniqueKeys);
+ if (1 << maxScale != uniqueKeys)
+ ++maxScale;
+ int avgScale = maxScale / (1 + k);
+ int scale = rnd.nextBiasedInt(0, avgScale, maxScale);
+ keyDomains[k] = rnd.nextInt(1 << scale, Math.min(uniqueKeys, 2 << scale));
+ uniqueKeys = (uniqueKeys + keyDomains[k] - 1) / keyDomains[k];
+ }
+ keyDomains[0] = uniqueKeys;
+ int[] valDomains = new int[valCount];
+ Arrays.fill(valDomains, Integer.MAX_VALUE);
+ int queryCount = rnd.nextInt(1, Math.max(2, Math.min(1000, rowCount)));
+ testOne(rnd, keyDomains, valDomains, rowCount, queryCount);
+ }
+
+ private void testOne(RandomSource rnd, int[] keyDomains, int[] valDomains, int rowCount, int queryLoops)
+ {
+ {
+ int uniqueKeys = 1;
+ for (int keyDomain : keyDomains)
+ uniqueKeys *= keyDomain;
+ Invariants.require(uniqueKeys >= 2 * rowCount);
+ }
+
+ int keyCount = keyDomains.length;
+ int valCount = valDomains.length;
+ String[] keyNames = IntStream.rangeClosed(1, keyCount).mapToObj(i -> "k" + i).toArray(String[]::new);
+ String[] valNames = IntStream.rangeClosed(1, valCount).mapToObj(i -> "v" + i).toArray(String[]::new);
+
+ int[][] allKeys = new int[rowCount][], allVals = new int[rowCount][];
+ {
+ TreeMap<int[], int[]> unique = new TreeMap<>(Arrays::compare);
+ for (int r = 0 ; r < rowCount ; ++r)
+ {
+ int[] vals = new int[valCount];
+ for (int v = 0 ; v < valCount ; ++v)
+ vals[v] = rnd.nextInt(valDomains[v]);
+
+ int[] keys = new int[keyCount];
+ do
+ {
+ for (int k = 0 ; k < keyCount ; ++k)
+ keys[k] = rnd.nextInt(keyDomains[k]);
+ }
+ while (null != unique.putIfAbsent(keys, vals));
+ }
+
+ int count = 0;
+ for (Map.Entry<int[], int[]> e : unique.entrySet())
+ {
+ allKeys[count] = e.getKey();
+ allVals[count] = e.getValue();
+ ++count;
+ }
+ }
+
+ class TestTable extends AbstractLazyVirtualTable
+ {
+ final int[][] copyOfKeys = new int[rowCount][], copyOfVals = new int[rowCount][];
+ TestTable(Sorted sorted, Sorted sortedByPartitionKey)
+ {
+ super(LazyVirtualTableTest.metadata(keyCount, valCount, sorted + "_" + sortedByPartitionKey), FAIL, sorted, sortedByPartitionKey);
+ IntUnaryOperator index;
+ switch (sorted)
+ {
+ default: throw new UnhandledEnum(sorted);
+ case SORTED: throw new UnsupportedOperationException(); // TODO (desired): test this
+ case ASC:
+ index = i -> i;
+ break;
+ case UNSORTED:
+ int[] indexes = IntStream.range(0, rowCount).toArray();
+ index = i -> indexes[i];
+ switch (sortedByPartitionKey)
+ {
+ default: throw new UnhandledEnum(sortedByPartitionKey);
+ case DESC: case SORTED: throw new UnsupportedOperationException(); // TODO (desired): test this
+ case UNSORTED:
+ shuffle(indexes, 0, indexes.length, rnd);
+ break;
+
+ case ASC:
+ int prev = 0;
+ for (int i = 1; i < rowCount ; ++i)
+ {
+ if (allKeys[prev][0] != allKeys[i][0])
+ {
+ shuffle(indexes, prev, i, rnd);
+ prev = i;
+ }
+ }
+ shuffle(indexes, prev, indexes.length, rnd);
+ }
+ break;
+ case DESC:
+ index = i -> rowCount - (i + 1);
+ break;
+ }
+ for (int i = 0 ; i < rowCount ; ++i)
+ {
+ copyOfKeys[i] = allKeys[index.applyAsInt(i)];
+ copyOfVals[i] = allVals[index.applyAsInt(i)];
+ }
+ }
+
+ @Override
+ protected void collect(PartitionsCollector collector)
+ {
+ for (int i = 0 ; i < copyOfKeys.length ; ++i)
+ {
+ int[] keys = copyOfKeys[i];
+ int[] vals = copyOfVals[i];
+ Object[] ckeys = new Object[keyCount - 1];
+ for (int k = 1; k < keyCount ; ++k)
+ ckeys[k - 1] = keys[k];
+ collector.partition(keys[0])
+ .collect(rows -> rows.add(ckeys).lazyCollect(cols -> {
+ for (int j = 0 ; j < vals.length ; ++j)
+ cols.add(valNames[j], vals[j]);
+ }));
+ }
+ }
+ }
+
+ List<VirtualTable> tables = new ArrayList<>();
+ tables.add(new TestTable(ASC, ASC));
+ tables.add(new TestTable(DESC, DESC));
+ tables.add(new TestTable(UNSORTED, ASC));
+ tables.add(new TestTable(UNSORTED, UNSORTED));
+ VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KEYSPACE, tables));
+
+ LargeBitSet matchingRows = new LargeBitSet(rowCount);
+ int fieldCount = keyCount + valCount;
+ LargeBitSet filteringKeys = new LargeBitSet(fieldCount);
+ StringBuilder where = new StringBuilder();
+ for (int q = 0 ; q < queryLoops ; q++)
+ {
+ matchingRows.setRange(0, rowCount);
+ filteringKeys.setRange(0, fieldCount);
+ where.setLength(0);
+ do
+ {
+ int f;
+ do
+ {
+ int min = filteringKeys.nextSetBit(0);
+ f = rnd.nextInt(min, fieldCount);
+ } while (!filteringKeys.unset(f));
+
+ int val;
+ if (rowCount == 0 || rnd.nextBoolean())
+ {
+ val = rnd.nextInt(f < keyCount ? keyDomains[f] : valDomains[f - keyCount]);
+ }
+ else
+ {
+ int i = rnd.nextInt(rowCount);
+ val = f < keyCount ? allKeys[i][f] : allVals[i][f - keyCount];
+ }
+ Cmp cmp = rnd.pick(CMPS);
+ if (where.length() > 0)
+ where.append(" AND ");
+
+ where.append(f < keyCount ? keyNames[f] : valNames[f - keyCount]);
+ where.append(' ');
+ where.append(cmp.str);
+ where.append(' ');
+ where.append(val);
+
+ IntBiPredicate pred = cmp.predicate();
+ for (int i = matchingRows.nextSetBit(0); i >= 0; i = matchingRows.nextSetBit(i + 1, -1))
+ {
+ int rowVal = f < keyCount ? allKeys[i][f] : allVals[i][f - keyCount];
+ if (!pred.test(rowVal, val))
+ matchingRows.unset(i);
+ }
+
+ int matchCount = matchingRows.getSetBitCount();
+ int limit = rnd.nextBoolean() ? 0 : rnd.nextInt(1, Math.max(2, 2 * matchCount));
+ for (VirtualTable table : tables)
+ {
+ UntypedResultSet results = execute("select * from " + table.metadata() + " where " + where + (limit <= 0 ? "" : " LIMIT " + limit));
+
+ int i = -1;
+ for (UntypedResultSet.Row row : results)
+ {
+ i = matchingRows.nextSetBit(i + 1);
+
+ for (int k = 0 ; k < keyCount ; ++k)
+ Assertions.assertThat(allKeys[i][k]).isEqualTo(row.getInt(keyNames[k]));
+
+ for (int v = 0 ; v < valCount ; ++v)
+ Assertions.assertThat(allVals[i][v]).isEqualTo(row.getInt(valNames[v]));
+ }
+
+ i = matchingRows.nextSetBit(i + 1);
+ if (limit >= matchCount || limit == 0) Assertions.assertThat(-1).isEqualTo(i);
+ else Assertions.assertThat(i).isNotEqualTo(-1);
+ Assertions.assertThat(results.size()).isEqualTo(limit == 0 ? matchCount : Math.min(matchCount, limit));
+ }
+ } while (matchingRows.getSetBitCount() > 0 && filteringKeys.getSetBitCount() > 0);
+ }
+
+ VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KEYSPACE, tables));
+ }
+
+ enum Cmp
+ {
+ LT("<"), LE("<="), NE("!="), EQ("="), GE(">="), GT(">");
+
+ static final Cmp[] CMPS = values();
+ final String str;
+
+ Cmp(String str)
+ {
+ this.str = str;
+ }
+
+ IntBiPredicate predicate()
+ {
+ switch (this)
+ {
+ default: throw new UnhandledEnum(this);
+ case LT: return (a, b) -> a < b;
+ case LE: return (a, b) -> a <= b;
+ case NE: return (a, b) -> a != b;
+ case EQ: return (a, b) -> a == b;
+ case GE: return (a, b) -> a >= b;
+ case GT: return (a, b) -> a > b;
+ }
+ }
+ }
+
+ private static TableMetadata metadata(int keys, int vals, String name)
+ {
+ return CreateTableStatement.parse(createTable(keys, vals, name), KEYSPACE)
+ .comment("")
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new LocalPartitioner(Int32Type.instance))
+ .build();
+ }
+
+ private static String createTable(int keys, int vals, String name)
+ {
+ StringBuilder sb = new StringBuilder("CREATE TABLE ");
+ sb.append(name).append(" (");
+ for (int i = 1 ; i <= keys ; ++i)
+ sb.append('k').append(i).append(" int, ");
+ for (int i = 1 ; i <= vals ; ++i)
+ sb.append('v').append(i).append(" int, ");
+ sb.append(" PRIMARY KEY (");
+ for (int i = 1 ; i <= keys ; ++i)
+ {
+ if (i > 1) sb.append(", ");
+ sb.append('k').append(i);
+ }
+ sb.append("))");
+ return sb.toString();
+ }
+
+ private static void shuffle(int[] array, int start, int end, RandomSource rnd)
+ {
+ while (end - start > 1)
+ {
+ int i = start + rnd.nextInt(end - start);
+ int tmp = array[start];
+ array[start] = array[i];
+ array[i] = tmp;
+ ++start;
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/db/virtual/LocalRepairTablesTest.java b/test/unit/org/apache/cassandra/db/virtual/LocalRepairTablesTest.java
index 5aa94cc666..75d335dafc 100644
--- a/test/unit/org/apache/cassandra/db/virtual/LocalRepairTablesTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/LocalRepairTablesTest.java
@@ -269,7 +269,7 @@ public class LocalRepairTablesTest extends CQLTester
private static RepairCoordinator.NeighborsAndRanges neighbors()
{
- return new RepairCoordinator.NeighborsAndRanges(false, ADDRESSES, ImmutableList.of(COMMON_RANGE));
+ return new RepairCoordinator.NeighborsAndRanges(false, false, ADDRESSES, ImmutableList.of(COMMON_RANGE));
}
private static Range<Token> range(long a, long b)
diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
index 9e6d406570..aa37658fc6 100644
--- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
@@ -521,7 +521,8 @@ public class RouteIndexTest extends CQLTester
AccordService.MetadataChangeListener.instance.resetForTesting(metadata);
NodeId tcmNodeId = metadata.myNodeId();
AccordService.unsafeSetNewAccordService(null);
- return AccordService.startup(tcmNodeId);
+ AccordService.localStartup(tcmNodeId);
+ return AccordService.distributedStartup();
}
TxnId nextTxnId(Domain domain)
diff --git a/test/unit/org/apache/cassandra/repair/NeighborsAndRangesTest.java b/test/unit/org/apache/cassandra/repair/NeighborsAndRangesTest.java
index d5c578d730..e39f0d2281 100644
--- a/test/unit/org/apache/cassandra/repair/NeighborsAndRangesTest.java
+++ b/test/unit/org/apache/cassandra/repair/NeighborsAndRangesTest.java
@@ -40,9 +40,9 @@ public class NeighborsAndRangesTest extends AbstractRepairTest
public void filterCommonIncrementalRangesNotForced()
{
CommonRange cr = new CommonRange(PARTICIPANTS, Collections.emptySet(), ALL_RANGES);
- NeighborsAndRanges nr = new NeighborsAndRanges(false, PARTICIPANTS, Collections.singletonList(cr));
+ NeighborsAndRanges nr = new NeighborsAndRanges(false, false, PARTICIPANTS, Collections.singletonList(cr));
List<CommonRange> expected = Lists.newArrayList(cr);
- List<CommonRange> actual = nr.filterCommonRanges(null, null);
+ List<CommonRange> actual = nr.filterCommonRanges(null, null).commonRanges;
Assert.assertEquals(expected, actual);
}
@@ -59,8 +59,8 @@ public class NeighborsAndRangesTest extends AbstractRepairTest
new CommonRange(Sets.newHashSet(PARTICIPANT2, PARTICIPANT3), Collections.emptySet(), Sets.newHashSet(RANGE3), true),
new CommonRange(Sets.newHashSet(PARTICIPANT2, PARTICIPANT3), Collections.emptySet(), Sets.newHashSet(RANGE2), false));
- NeighborsAndRanges nr = new NeighborsAndRanges(true, liveEndpoints, initial);
- List<CommonRange> actual = nr.filterCommonRanges(null, null);
+ NeighborsAndRanges nr = new NeighborsAndRanges(true, false, liveEndpoints, initial);
+ List<CommonRange> actual = nr.filterCommonRanges(null, null).commonRanges;
Assert.assertEquals(expected, actual);
}
diff --git a/test/unit/org/apache/cassandra/repair/RepairJobTest.java b/test/unit/org/apache/cassandra/repair/RepairJobTest.java
index bea5cef88a..88a8de34c9 100644
--- a/test/unit/org/apache/cassandra/repair/RepairJobTest.java
+++ b/test/unit/org/apache/cassandra/repair/RepairJobTest.java
@@ -128,8 +128,8 @@ public class RepairJobTest
boolean dontPurgeTombstones, boolean repairAccord, String... cfnames)
{
super(SharedContext.Global.instance, new Scheduler.NoopScheduler(),
- parentRepairSession, commonRange, excludedDeadNodes, keyspace, parallelismDegree, isIncremental, pullRepair,
- previewKind, optimiseStreams, repairData, repairPaxos, dontPurgeTombstones, repairAccord, cfnames);
+ parentRepairSession, commonRange, excludedDeadNodes, keyspace, parallelismDegree, false, isIncremental, pullRepair,
+ previewKind, optimiseStreams, repairData, repairPaxos, dontPurgeTombstones, repairAccord, false, cfnames);
}
@Override
diff --git a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java
index c59be70c7b..371c41b64f 100644
--- a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java
+++ b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java
@@ -66,8 +66,8 @@ public class RepairSessionTest
RepairSession session = new RepairSession(SharedContext.Global.instance, new Scheduler.NoopScheduler(), parentSessionId,
new CommonRange(endpoints, Collections.emptySet(), Arrays.asList(repairRange)),
false, "Keyspace1", RepairParallelism.SEQUENTIAL,
- false, false, PreviewKind.NONE, false,
- false, false, false, false, "Standard1");
+ false, false, false, PreviewKind.NONE, false,
+ false, false, false, false, false, "Standard1");
// perform convict
session.convict(remote, Double.MAX_VALUE);
diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairParameterizedTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairParameterizedTest.java
index 8e6a559d33..3d98e61fcf 100644
--- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairParameterizedTest.java
+++ b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairParameterizedTest.java
@@ -37,8 +37,6 @@ import org.apache.cassandra.cql3.statements.schema.TableAttributes;
import org.apache.cassandra.repair.RepairCoordinator;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.LocalStrategy;
-import org.apache.cassandra.repair.RepairParallelism;
-import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.service.StorageService;
@@ -62,7 +60,6 @@ import org.apache.cassandra.metrics.AutoRepairMetrics;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.AutoRepairService;
-import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
@@ -811,9 +808,6 @@ public class AutoRepairParameterizedTest extends CQLTester
@Test
public void testSchedulerIgnoresErrorsFromUnrelatedRepairRunables()
{
- RepairOption options = new RepairOption(RepairParallelism.PARALLEL, true, repairType == AutoRepairConfig.RepairType.INCREMENTAL, false,
- AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), Collections.emptySet(),
- false, false, PreviewKind.NONE, false, true, true, false, false, false);
AutoRepairState repairState = AutoRepair.instance.repairStates.get(repairType);
AutoRepairState spyState = spy(repairState);
AtomicReference<AutoRepair.RepairProgressListener> failingListener = new AtomicReference<>();
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCacheTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCacheTest.java
index c23c0c0601..02de1ad641 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCacheTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCacheTest.java
@@ -280,7 +280,8 @@ public class AccordCacheTest
public void testEvictionOnAcquire()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
- AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
+ NoOpLock lock = new NoOpLock();
+ AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 5);
@@ -297,6 +298,7 @@ public class AccordCacheTest
testLoad(executor, safeString, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(safeString.key()));
instance.release(safeString, null);
+ cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 0, 5, nodeSize(1) * 5);
@@ -305,6 +307,7 @@ public class AccordCacheTest
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
SafeString safeString = instance.acquire("5");
+ cache.tryShrinkOrEvict(lock);
Assert.assertTrue(instance.isReferenced(safeString.key()));
// since it's not loaded, only the node size is counted here
@@ -327,7 +330,8 @@ public class AccordCacheTest
public void testEvictionOnRelease()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
- AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
+ NoOpLock lock = new NoOpLock();
+ AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 4);
@@ -343,6 +347,7 @@ public class AccordCacheTest
items[i] = safeString;
testLoad(executor, safeString, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(safeString.key()));
+ cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 5, 5, nodeSize(1) * 5);
@@ -351,12 +356,14 @@ public class AccordCacheTest
Assert.assertNull(cache.tail());
instance.release(items[2], null);
+ cache.tryShrinkOrEvict(lock);
assertCacheState(cache, 4, 4, nodeSize(1) * 4);
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
Assert.assertNull(cache.head());
Assert.assertNull(cache.tail());
instance.release(items[4], null);
+ cache.tryShrinkOrEvict(lock);
assertCacheState(cache, 3, 4, nodeSize(1) * 4);
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
Assert.assertSame(items[4].global, cache.head());
@@ -401,7 +408,8 @@ public class AccordCacheTest
public void evictionBlockedOnSaving()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
- AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
+ NoOpLock lock = new NoOpLock();
+ AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 3 + nodeSize(3));
@@ -422,6 +430,7 @@ public class AccordCacheTest
testLoad(executor, item, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(item.key()));
instance.release(item, null);
+ cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 0, 4, nodeSize(1) * 3 + nodeSize(2));
@@ -430,6 +439,7 @@ public class AccordCacheTest
// force cache eviction
instance.acquire(Integer.toString(0));
cache.setCapacity(0);
+ cache.tryShrinkOrEvict(lock);
// all should have been evicted except 0
assertCacheState(cache, 1, 1, nodeSize(2));
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
index f43ddb66e4..931f3511c5 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
@@ -103,7 +103,7 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
- Route<?> route = fullRoute.slice(fullRange(txn));
+ Route<?> route = fullRoute.overlapping(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, null, false, fullRoute);
@@ -195,7 +195,7 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
- Route<?> route = fullRoute.slice(fullRange(txn));
+ Route<?> route = fullRoute.overlapping(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, null, false, fullRoute);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
index 0c107e0f7c..26075bdf84 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
@@ -28,7 +28,6 @@ import accord.messages.ReadData;
import accord.messages.ReadData.CommitOrReadNack;
import accord.topology.TopologyUtils;
import org.apache.cassandra.service.accord.AccordFetchCoordinator.AccordFetchRequest;
-import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.AccordTimeService;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@@ -72,7 +71,7 @@ public class AccordMessageSinkTest
DatabaseDescriptor.clientInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ClusterMetadataService.initializeForClients();
- sink = new AccordMessageSink(Mockito.mock(AccordAgent.class), messaging, mapping, new RequestCallbacks(new AccordTimeService()));
+ sink = new AccordMessageSink(messaging, mapping, new RequestCallbacks(new AccordTimeService()));
}
@Test
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
index 520e96f35a..8407f0e74b 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
@@ -195,7 +195,7 @@ public class AccordTaskTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
- route.slice(ranges);
+ route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
@@ -205,10 +205,15 @@ public class AccordTaskTest
}));
// clear cache
- try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
+ long cacheSize;
+ try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches())
{
- long cacheSize = cache.global.capacity();
+ cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
+ }
+
+ try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches())
+ {
cache.global.setCapacity(cacheSize);
}
@@ -238,7 +243,7 @@ public class AccordTaskTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
- Route<?> partialRoute = route.slice(ranges);
+ Route<?> partialRoute = route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
@@ -250,10 +255,14 @@ public class AccordTaskTest
}));
// clear cache
+ long cacheSize;
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
- long cacheSize = cache.global.capacity();
+ cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
+ }
+ try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
+ {
cache.global.setCapacity(cacheSize);
}
@@ -280,57 +289,58 @@ public class AccordTaskTest
.withExamples(50)
.forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 2))
.check((rs, ids) -> {
- before(); // truncate tables
- Participants<RoutingKey> participants = keys.toParticipants();
- assertNoReferences(commandStore, ids, participants);
- createCommand(commandStore, rs, ids);
- awaitDone(commandStore, ids, participants);
- assertNoReferences(commandStore, ids, participants);
+ before(); // truncate tables
- PreLoadContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
- Consumer<SafeCommandStore> consumer = Mockito.mock(Consumer.class);
+ Participants<RoutingKey> participants = keys.toParticipants();
+ assertNoReferences(commandStore, ids, participants);
+ createCommand(commandStore, rs, ids);
+ awaitDone(commandStore, ids, participants);
+ assertNoReferences(commandStore, ids, participants);
- Map<TxnId, Boolean> failed = selectFailedTxn(rs, ids);
- try (ExclusiveGlobalCaches caches = commandStore.executor().lockCaches())
- {
- caches.commands.unsafeSetLoadFunction((s, txnId) ->
+ PreLoadContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
+ Consumer<SafeCommandStore> consumer = Mockito.mock(Consumer.class);
+
+ Map<TxnId, Boolean> failed = selectFailedTxn(rs, ids);
+ try (ExclusiveGlobalCaches caches = commandStore.executor().lockCaches())
{
- logger.info("Attempting to load {}; expected to fail? {}", txnId, failed.get(txnId));
- if (!failed.get(txnId))
- return commandStore.loadCommand(txnId);
- throw new NullPointerException("txn_id " + txnId);
- });
- }
- AccordTask<Void> o1 = AccordTask.create(commandStore, ctx, consumer);
- AssertionUtils.assertThatThrownBy(() -> getBlocking(o1.chain()))
- .hasRootCause()
- .isInstanceOf(NullPointerException.class)
- .hasNoSuppressedExceptions();
+ caches.commands.unsafeSetLoadFunction((s, txnId) ->
+ {
+ logger.info("Attempting to load {}; expected to fail? {}", txnId, failed.get(txnId));
+ if (!failed.get(txnId))
+ return commandStore.loadCommand(txnId);
+ throw new NullPointerException("txn_id " + txnId);
+ });
+ }
+ AccordTask<Void> o1 = AccordTask.create(commandStore, ctx, consumer);
+ AssertionUtils.assertThatThrownBy(() -> getBlocking(o1.chain()))
+ .hasRootCause()
+ .isInstanceOf(NullPointerException.class)
+ .hasNoSuppressedExceptions();
- Mockito.verifyNoInteractions(consumer);
+ Mockito.verifyNoInteractions(consumer);
- assertNoReferences(commandStore, ids, participants);
- // the first failed load causes the whole operation to fail, so some ids may still be pending
- // to make sure the next operation does not see a PENDING that will fail, wait for all loads to complete
- awaitDone(commandStore, ids, participants);
+ assertNoReferences(commandStore, ids, participants);
+ // the first failed load causes the whole operation to fail, so some ids may still be pending
+ // to make sure the next operation does not see a PENDING that will fail, wait for all loads to complete
+ awaitDone(commandStore, ids, participants);
- // can we recover?
- try (ExclusiveGlobalCaches caches = commandStore.executor().lockCaches())
- {
- caches.commands.unsafeSetLoadFunction((s, txnId) -> {
- Command cmd = commandStore.loadCommand(txnId);
- return cmd;
- });
- }
- AccordTask<Void> o2 = AccordTask.create(commandStore, ctx, store -> {
- ids.forEach(id -> {
- store.ifInitialised(id).readyToExecute(store);
+ // can we recover?
+ try (ExclusiveGlobalCaches caches = commandStore.executor().lockCaches())
+ {
+ caches.commands.unsafeSetLoadFunction((s, txnId) -> {
+ Command cmd = commandStore.loadCommand(txnId);
+ return cmd;
+ });
+ }
+ AccordTask<Void> o2 = AccordTask.create(commandStore, ctx, store -> {
+ ids.forEach(id -> {
+ store.ifInitialised(id).readyToExecute(store);
+ });
});
- });
- getBlocking(o2.chain());
- awaitDone(commandStore, ids, participants);
- assertNoReferences(commandStore, ids, participants);
+ getBlocking(o2.chain());
+ awaitDone(commandStore, ids, participants);
+ assertNoReferences(commandStore, ids, participants);
});
}
diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
index 63051f72f8..b94237befc 100644
--- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
@@ -392,7 +392,7 @@ public class EpochSyncTest
.isFalse();
// validate topology manager
- Ranges ranges = tm.active().globalForEpoch(epoch).ranges().mergeTouching();
+ Ranges ranges = tm.active().getKnown(epoch).global().ranges().mergeTouching();
Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching();
Assertions.assertThat(actual)
.describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual))
@@ -404,7 +404,7 @@ public class EpochSyncTest
continue;
Assertions.assertThat(tm.active().hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
- Topology topology = tm.active().globalForEpoch(epoch);
+ Topology topology = tm.active().getKnown(epoch).global();
Ranges ranges = topology.ranges().mergeTouching();
Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching();
// TopologyManager defines syncComplete for an epoch as (epoch - 1).syncComplete. This means that an epoch has reached quorum, but will still miss ranges as previous epochs have not
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
index fceec176bc..0cbf4ae650 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
@@ -159,7 +159,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
this.topologies = new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, topology);
Ranges ranges = topology.ranges();
if (tableId != null)
- ranges = ranges.slice(Ranges.of(TokenRange.create(TokenKey.min(tableId, getPartitioner()), TokenKey.max(tableId, getPartitioner()))));
+ ranges = ranges.overlapping(Ranges.of(TokenRange.create(TokenKey.min(tableId, getPartitioner()), TokenKey.max(tableId, getPartitioner()))));
CommandStores.RangesForEpoch rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), ranges);
updateHolder.add(topology.epoch(), rangesForEpoch, ranges);
@@ -258,10 +258,10 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
@Override
- public void onUncaughtException(Throwable t)
+ public void onException(Throwable t)
{
if (ignoreExceptions.test(t)) return;
- super.onUncaughtException(t);
+ super.onException(t);
}
};
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 7a348d6c45..ce9e9cdb13 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
@@ -662,8 +662,8 @@ public class CommandsForKeySerializerTest
@Override public void shutdown() { }
@Override public <T> AsyncChain<T> chain(Callable<T> call) { throw new UnsupportedOperationException(); }
@Override public OwnershipEventListener ownershipEvents() { return null; }
- @Override public void onUncaughtException(Throwable t) { throw new UnsupportedOperationException(); }
- @Override public void onCaughtException(Throwable t, String context) { throw new UnsupportedOperationException(); }
+ @Override public void onException(Throwable t) { throw new UnsupportedOperationException(); }
+ @Override public void onException(Throwable t, String context) { throw new UnsupportedOperationException(); }
@Override public boolean rejectPreAccept(TimeService time, TxnId txnId) { throw new UnsupportedOperationException(); }
@Override public long cfkHlcPruneDelta() { return 0; }
@Override public int cfkPruneInterval() { return 0; }
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java
index 6dc45af8bb..4335085433 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java
@@ -181,7 +181,7 @@ public class KeySerializersTest
int count = superset.isEmpty() ? 0 : rs.nextInt(superset.size());
Participants<?> subset = selectSubset(rs, count, superset);
if (superset instanceof Route<?> && (!changeType || rs.nextBoolean()))
- return superset.intersecting(subset);
+ return superset.overlapping(subset);
return subset;
}
diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java
index 9edf8ea089..4c6827710d 100644
--- a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java
@@ -122,7 +122,7 @@ public class TxnUpdateTest
}
// slice the same key should return the same block
- TxnUpdate noUpdate = update.getTxnUpdate(k -> k.intersecting(update.keys()));
+ TxnUpdate noUpdate = update.getTxnUpdate(k -> k.overlapping(update.keys()));
for (int i = 0; i < update.blocks.size(); i++)
assertThat(noUpdate.blocks.get(i)).isSameAs(update.blocks.get(i));
@@ -131,7 +131,7 @@ public class TxnUpdateTest
int keyIndex = rs.nextInt(0, update.keys().size());
Key key = update.keys().get(keyIndex);
Keys singleKey = Keys.of(key);
- TxnUpdate singleKeyUpdate = update.getTxnUpdate(k -> k.intersecting(singleKey));
+ TxnUpdate singleKeyUpdate = update.getTxnUpdate(k -> k.overlapping(singleKey));
for (int i = 0; i < update.blocks.size(); i++)
{
var block = singleKeyUpdate.blocks.get(i);
@@ -162,7 +162,7 @@ public class TxnUpdateTest
for (int i = 0; i < update.keys().size(); i++)
{
int finalI = i;
- perKeyUpdate.add(update.getTxnUpdate(k -> k.intersecting(Keys.of(update.keys().get(finalI)))));
+ perKeyUpdate.add(update.getTxnUpdate(k -> k.overlapping(Keys.of(update.keys().get(finalI)))));
}
assertThat(update.merge(update)).isEqualTo(update); // merge with self produces self
diff --git a/test/unit/org/apache/cassandra/service/paxos/uncommitted/PaxosStateTrackerTest.java b/test/unit/org/apache/cassandra/service/paxos/uncommitted/PaxosStateTrackerTest.java
index 71d4877e57..3ef0d4e3f8 100644
--- a/test/unit/org/apache/cassandra/service/paxos/uncommitted/PaxosStateTrackerTest.java
+++ b/test/unit/org/apache/cassandra/service/paxos/uncommitted/PaxosStateTrackerTest.java
@@ -75,7 +75,7 @@ public class PaxosStateTrackerTest
{
SchemaLoader.prepareServer();
DatabaseDescriptor.setAccordTransactionsEnabled(false);
- AccordService.startup(null);
+ AccordService.localStartup(null);
ks = "coordinatorsessiontest";
cfm1 = TableMetadata.builder(ks, "tbl1").addPartitionKeyColumn("k", Int32Type.instance).addRegularColumn("v", Int32Type.instance).build();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org