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/06/19 11:40:59 UTC
(cassandra) branch trunk updated (7140641927 -> 1350082a13)
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 7140641927 Avoid re-serializing all-equivalent topologies
omit b6e924f8e5 Erase should consistently erase all content Also Fix: - slowCoordinatorDelay should bound its start point, and log more detail if its expectations are breached - Erased SaveStatus can be reported for all queried owned participants on a replica (the saved participants have been erased)
new 7ef2ccd4a1 Erase should consistently erase all content Also Fix: - slowCoordinatorDelay should bound its start point, and log more detail if its expectations are breached - Erased SaveStatus can be reported for all queried owned participants on a replica (the saved participants have been erased)
new 1350082a13 Avoid re-serializing all-equivalent topologies
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 (7140641927)
\
N -- N -- N refs/heads/trunk (1350082a13)
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 2 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:
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java | 2 +-
.../cassandra/service/accord/journal/AccordTopologyUpdate.java | 2 ++
.../cassandra/distributed/test/accord/AccordProgressLogTest.java | 6 +++---
.../cassandra/service/accord/journal/AccordTopologyUpdateTest.java | 1 +
4 files changed, 7 insertions(+), 4 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 02/02: Avoid re-serializing all-equivalent topologies
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 1350082a13f2ba534750704d340d4b16496f9634
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Wed Jun 18 21:37:34 2025 +0200
Avoid re-serializing all-equivalent topologies
patch by Alex Petrov; reviewed by Benedict for CASSANDRA-20721
---
modules/accord | 2 +-
.../accord/journal/AccordTopologyUpdate.java | 111 +++++++++++++++++++--
.../fuzz/topology/IdenticalTopologyTest.java | 97 ++++++++++++++++++
.../accord/journal/AccordTopologyUpdateTest.java | 3 +-
4 files changed, 201 insertions(+), 12 deletions(-)
diff --git a/modules/accord b/modules/accord
index 2b80b0233f..07ba9c85d0 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 2b80b0233f96464430058d70f7398df11066c3a6
+Subproject commit 07ba9c85d083b586ff5c2382ef747b0fe71a9c47
diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java
index a01580363f..940c9090a5 100644
--- a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java
+++ b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java
@@ -51,6 +51,7 @@ public interface AccordTopologyUpdate
void applyTo(TopologyImage accumulator);
long epoch();
+ Journal.TopologyUpdate getUpdate();
static AccordTopologyUpdate newTopology(Journal.TopologyUpdate update)
{
return new NewTopology(update);
@@ -161,9 +162,25 @@ public interface AccordTopologyUpdate
switch (t.kind())
{
case NewTopology:
+ {
TopologyUpdateSerializer.instance.serialize(((NewTopology) t).update, out);
break;
- case Topologies:
+ }
+ case NoOp:
+ {
+ TopologyImage image = (TopologyImage) t;
+ Invariants.require(image.update == null);
+ if (image.syncStatus == null)
+ out.writeByte(Byte.MAX_VALUE);
+ else
+ out.writeByte(image.syncStatus.ordinal());
+
+ KeySerializers.ranges.serialize(image.closed, out);
+ KeySerializers.ranges.serialize(image.retired, out);
+ break;
+ }
+ case TopologyImage:
+ {
TopologyImage image = (TopologyImage) t;
out.writeBoolean(image.update != null);
@@ -177,6 +194,7 @@ public interface AccordTopologyUpdate
KeySerializers.ranges.serialize(image.closed, out);
KeySerializers.ranges.serialize(image.retired, out);
break;
+ }
default:
throw new UnhandledEnum(t.kind());
}
@@ -191,9 +209,20 @@ public interface AccordTopologyUpdate
{
case NewTopology:
return new NewTopology(TopologyUpdateSerializer.instance.deserialize(in));
- case Topologies:
+ case NoOp:
{
- TopologyImage image = new TopologyImage(epoch);
+ TopologyImage image = new TopologyImage(epoch, Kind.NoOp);
+ byte syncStateByte = in.readByte();
+ if (syncStateByte != Byte.MAX_VALUE)
+ image.syncStatus = AccordConfigurationService.SyncStatus.values()[syncStateByte];
+
+ image.closed = KeySerializers.ranges.deserialize(in);
+ image.retired = KeySerializers.ranges.deserialize(in);
+ return image;
+ }
+ case TopologyImage:
+ {
+ TopologyImage image = new TopologyImage(epoch, Kind.TopologyImage);
if (in.readBoolean())
image.update = TopologyUpdateSerializer.instance.deserialize(in);
@@ -221,7 +250,19 @@ public interface AccordTopologyUpdate
case NewTopology:
size += TopologyUpdateSerializer.instance.serializedSize(((NewTopology) t).update);
break;
- case Topologies:
+ case NoOp:
+ {
+ TopologyImage image = (TopologyImage) t;
+ Invariants.require(image.update == null);
+
+ size += Byte.BYTES;
+
+ size += KeySerializers.ranges.serializedSize(image.closed);
+ size += KeySerializers.ranges.serializedSize(image.retired);
+ break;
+ }
+ case TopologyImage:
+ {
TopologyImage image = (TopologyImage) t;
size += TypeSizes.sizeof(image.update != null);
@@ -233,6 +274,7 @@ public interface AccordTopologyUpdate
size += KeySerializers.ranges.serializedSize(image.closed);
size += KeySerializers.ranges.serializedSize(image.retired);
break;
+ }
default:
throw new UnhandledEnum(t.kind());
}
@@ -242,8 +284,12 @@ public interface AccordTopologyUpdate
enum Kind
{
+ // New Topology, written to journal when the node first learned about it
NewTopology,
- Topologies
+ // Used when accumulating state during compaction or replay
+ TopologyImage,
+ // Effectively unchanged topology
+ NoOp
}
class ImmutableTopoloyImage extends Journal.TopologyUpdate
@@ -263,10 +309,30 @@ public interface AccordTopologyUpdate
private Ranges retired = Ranges.EMPTY;
private final long epoch;
+ private final Kind kind;
- public TopologyImage(long epoch)
+ public TopologyImage(long epoch, Kind kind)
{
+ Invariants.require(kind != Kind.NewTopology);
this.epoch = epoch;
+ this.kind = kind;
+ }
+
+ public TopologyImage asImage(Journal.TopologyUpdate update)
+ {
+ TopologyImage image = new TopologyImage(epoch, Kind.TopologyImage);
+ image.update = update.cloneWithEquivalentEpoch(epoch);
+ image.closed = closed;
+ image.retired = retired;
+ return image;
+ }
+
+ public TopologyImage asNoOp()
+ {
+ TopologyImage image = new TopologyImage(epoch, Kind.NoOp);
+ image.closed = closed;
+ image.retired = retired;
+ return image;
}
@Override
@@ -275,10 +341,16 @@ public interface AccordTopologyUpdate
return this.epoch;
}
+ @Override
+ public Journal.TopologyUpdate getUpdate()
+ {
+ return update;
+ }
+
@Override
public Kind kind()
{
- return Kind.Topologies;
+ return kind;
}
@Override
@@ -327,6 +399,12 @@ public interface AccordTopologyUpdate
return this.epoch;
}
+ @Override
+ public Journal.TopologyUpdate getUpdate()
+ {
+ return update;
+ }
+
@Override
public Kind kind()
{
@@ -385,7 +463,7 @@ public interface AccordTopologyUpdate
@Override
protected NavigableMap<Long, TopologyImage> accumulate(NavigableMap<Long, TopologyImage> allEpochs, AccordTopologyUpdate update)
{
- update.applyTo(allEpochs.computeIfAbsent(update.epoch(), v -> new TopologyImage(update.epoch())));
+ update.applyTo(allEpochs.computeIfAbsent(update.epoch(), v -> new TopologyImage(update.epoch(), Kind.TopologyImage)));
return allEpochs;
}
}
@@ -434,8 +512,16 @@ public interface AccordTopologyUpdate
public void reserialize(JournalKey key, Accumulator from, DataOutputPlus out, Version version) throws IOException
{
out.writeUnsignedVInt32(from.get().size());
+ Journal.TopologyUpdate prev = null;
for (TopologyImage value : from.get().values())
+ {
+ Journal.TopologyUpdate tmp = value.update;
+ if (prev != null && value.update.isEquivalent(prev))
+ value = value.asNoOp();
+
+ prev = tmp;
Serializer.instance.serialize(value, out);
+ }
}
@Override
@@ -443,13 +529,18 @@ public interface AccordTopologyUpdate
{
long minEpoch = this.minEpoch.epoch();
int count = in.readUnsignedVInt32();
+ AccordTopologyUpdate prev = null;
while (--count >= 0)
{
AccordTopologyUpdate update = Serializer.instance.deserialize(in);
+ if (update.kind() == Kind.NoOp)
+ {
+ Invariants.require(prev != null);
+ update = ((TopologyImage) update).asImage(prev.getUpdate());
+ }
if (update.epoch() >= minEpoch)
into.update(update);
- else
- return;
+ prev = update;
}
}
}
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java
new file mode 100644
index 0000000000..b35954735e
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.fuzz.topology;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.test.log.FuzzTestBase;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
+import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
+import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
+import org.apache.cassandra.harry.gen.Generator;
+import org.apache.cassandra.harry.gen.SchemaGenerators;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.service.accord.AccordKeyspace;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+
+import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+
+public class IdenticalTopologyTest extends FuzzTestBase
+{
+ private static final int POPULATION = 1000;
+
+ @Test
+ public void identicalTopologyTest() throws Throwable
+ {
+ try (Cluster cluster = init(builder().withNodes(1)
+ .withConfig(cfg -> cfg.set("accord.journal.compaction_period", "600s") // deliberately high
+ .set("max_value_size", "1MiB")) // deliberately low
+ .start(),
+ 1))
+ {
+ withRandom(rng -> {
+ cluster.get(1).runOnInstance(() -> {
+ Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).disableAutoCompaction();
+ });
+
+ Generator<SchemaSpec> schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, "bootstrap_fuzz", POPULATION,
+ SchemaSpec.optionsBuilder()
+ .addWriteTimestamps(false)
+ .withTransactionalMode(TransactionalMode.full));
+
+ SchemaSpec schema = schemaGen.generate(rng);
+ cluster.schemaChange(schema.compile());
+ HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ hb -> InJvmDTestVisitExecutor.builder()
+ .consistencyLevel(ConsistencyLevel.QUORUM)
+ .wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
+ .pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
+ .build(schema, hb, cluster));
+
+ for (int i = 0; i <= 100; i++)
+ {
+ cluster.schemaChange(String.format("ALTER TABLE %s.%s WITH comment = '%d';", schema.keyspace, schema.table, i));
+
+ for (int j = 0; j < 5; j++)
+ history.insert(j);
+ if (i % 20 == 0)
+ {
+ cluster.get(1).runOnInstance(() -> {
+ ((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty();
+ });
+ }
+ }
+ cluster.get(1).runOnInstance(() -> {
+ ((AccordService) AccordService.instance()).journal().compactor().run();
+ });
+ cluster.get(1).forceCompact("system_accord", "journal");
+ ClusterUtils.stopUnchecked(cluster.get(1));
+ cluster.get(1).startup();
+ });
+ }
+ }
+}
+
diff --git a/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java
index c9ed1c5af3..109aa3ed46 100644
--- a/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java
@@ -141,7 +141,8 @@ public class AccordTopologyUpdateTest
switch (kind)
{
case NewTopology: return new AccordTopologyUpdate.NewTopology(topologyUpdateGen.next(rs));
- case Topologies: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs));
+ case TopologyImage: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs), AccordTopologyUpdate.Kind.TopologyImage);
+ case NoOp: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs), AccordTopologyUpdate.Kind.NoOp);
default: throw new AssertionError("Unknown kind: " + kind);
}
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/02: Erase should consistently erase all content Also Fix: - slowCoordinatorDelay should bound its start point, and log more detail if its expectations are breached - Erased SaveStatus can be reported for all queried owned participants on a replica (the saved participants have been erased)
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 7ef2ccd4a1339a63d346523520fa2f9dae8352f5
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Jun 18 13:04:26 2025 +0100
Erase should consistently erase all content
Also Fix:
- slowCoordinatorDelay should bound its start point, and log more detail if its expectations are breached
- Erased SaveStatus can be reported for all queried owned participants on a replica (the saved participants have been erased)
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20722
---
modules/accord | 2 +-
.../db/compaction/CompactionIterator.java | 26 ++++++++++++++--------
.../cassandra/service/accord/AccordJournal.java | 10 ++++++++-
.../cassandra/service/accord/api/AccordAgent.java | 12 ++++++----
.../test/accord/AccordProgressLogTest.java | 6 ++---
5 files changed, 38 insertions(+), 18 deletions(-)
diff --git a/modules/accord b/modules/accord
index 1ce7122e2e..2b80b0233f 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 1ce7122e2e305f7510ec4c10c7587822c0549364
+Subproject commit 2b80b0233f96464430058d70f7398df11066c3a6
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 53b4d57f2e..e95b100a3f 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -35,6 +35,7 @@ import com.google.common.collect.Ordering;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.impl.CommandChange;
import accord.local.Cleanup;
import accord.local.DurableBefore;
import accord.local.RedundantBefore;
@@ -109,6 +110,7 @@ import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement;
import org.apache.cassandra.utils.TimeUUID;
+import static accord.local.Cleanup.ERASE;
import static accord.local.Cleanup.Input.PARTIAL;
import static accord.local.Cleanup.NO;
import static com.google.common.base.Preconditions.checkState;
@@ -117,6 +119,7 @@ import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
import static org.apache.cassandra.service.accord.AccordKeyspace.CFKAccessor;
+import static org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns.getJournalKey;
/**
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
@@ -880,7 +883,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
@Override
protected void beginPartition(UnfilteredRowIterator partition)
{
- key = AccordKeyspace.JournalColumns.getJournalKey(partition.partitionKey());
+ key = getJournalKey(partition.partitionKey());
if (compactor == null || compactor.serializer != key.type.serializer)
{
switch (key.type)
@@ -895,15 +898,12 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
compactor = new AccordMergingCompactor(key.type.serializer, userVersion);
}
}
- compactor.reset(key);
+ compactor.reset(key, partition);
}
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
- if (!partition.hasNext())
- return partition;
-
try
{
beginPartition(partition);
@@ -941,7 +941,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
this.serializer = serializer;
}
- abstract void reset(JournalKey key);
+ abstract void reset(JournalKey key, UnfilteredRowIterator partition);
abstract void collect(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException;
abstract UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException;
}
@@ -962,7 +962,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
@Override
- void reset(JournalKey key)
+ void reset(JournalKey key, UnfilteredRowIterator partition)
{
builder.reset(key);
lastDescriptor = -1;
@@ -1060,13 +1060,15 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
@Override
- void reset(JournalKey key)
+ void reset(JournalKey key, UnfilteredRowIterator partition)
{
mainBuilder.reset(key);
reuseEntries.addAll(entries);
for (int i = 0; i < entries.size() ; ++i)
entries.get(i).clear();
entries.clear();
+ if (!partition.partitionLevelDeletion().isLive())
+ mainBuilder.addCleanup(false, ERASE);
}
@Override
@@ -1103,7 +1105,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
case EXPUNGE:
return null;
case ERASE:
- return PartitionUpdate.fullPartitionDelete(AccordKeyspace.Journal, partitionKey, Long.MAX_VALUE, nowInSec).unfilteredIterator();
+ return erase(partitionKey);
case TRUNCATE:
case TRUNCATE_WITH_OUTCOME:
@@ -1136,8 +1138,14 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
return newVersion.build().unfilteredIterator();
}
+
+ private UnfilteredRowIterator erase(DecoratedKey partitionKey)
+ {
+ return PartitionUpdate.fullPartitionDelete(AccordKeyspace.Journal, partitionKey, Long.MAX_VALUE, nowInSec).unfilteredIterator();
+ }
}
+
private static class AbortableUnfilteredPartitionTransformation extends Transformation<UnfilteredRowIterator>
{
private final AbortableUnfilteredRowTransformation abortableIter;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index f98061f887..d7c507e34f 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -588,7 +588,15 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
ResultSerializers.result.serialize(command.result(), out);
break;
case CLEANUP:
- throw new IllegalStateException();
+ Cleanup cleanup;
+ switch (command.saveStatus())
+ {
+ default: throw new UnhandledEnum(command.saveStatus());
+ case Erased: cleanup = Cleanup.ERASE; break;
+ case Invalidated: cleanup = Cleanup.INVALIDATE; break;
+ }
+ out.writeByte(cleanup.ordinal());
+ break;
}
iterable = unsetIterable(field, iterable);
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 949f7bbf86..9f07e351ab 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -223,18 +223,22 @@ public class AccordAgent implements Agent
Command command = safeCommand.current();
Invariants.nonNull(command);
- Timestamp mostRecentAttempt = Timestamp.max(command.txnId(), command.promised());
RoutingKey homeKey = command.route().homeKey();
Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch());
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long oneSecond = SECONDS.toMicros(1L);
- long startTime = mostRecentAttempt.hlc() + recover(txnId).computeWait(retryCount, MICROSECONDS);
+ long mostRecentStart = Math.max(command.txnId().hlc(), command.promised().hlc());
+ long waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS);
+ long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
+ Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros);
+ long startTime = mostRecentStart + waitMicros;
+ if (startTime < nowMicros)
+ startTime = nowMicros + waitMicros/2;
startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), oneSecond, random);
- long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
long delayMicros = Math.max(1, startTime - nowMicros);
- Invariants.require(delayMicros < TimeUnit.HOURS.toMicros(1L));
+ Invariants.require(delayMicros < TimeUnit.HOURS.toMicros(1L), "unexpectedly long coordination recovery delay proposed: %d (start %d, now %d)", delayMicros, startTime, nowMicros, command.txnId(), command.promised());
return units.convert(delayMicros, MICROSECONDS);
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
index 6bbbce6a52..3f4163ddb2 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
@@ -46,7 +46,7 @@ public class AccordProgressLogTest extends TestBaseImpl
.withoutVNodes()
.withConfig(c -> c.with(Feature.NETWORK)
.set("accord.enabled", "true")
- .set("accord.recover_txn", "1s"))
+ .set("accord.recover_txn", "100ms"))
.start()))
{
cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}");
@@ -59,7 +59,7 @@ public class AccordProgressLogTest extends TestBaseImpl
AtomicLong recoveryStartedAt = new AtomicLong();
Semaphore waitForRecovery = new Semaphore(0);
IMessageFilters.Filter recovery = cluster.filters().outbound().messagesMatching((from, to, message) -> {
- if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_RSP.id)
+ if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_REQ.id)
{
recoveryStartedAt.compareAndSet(0, System.nanoTime());
waitForRecovery.release();
@@ -75,7 +75,7 @@ public class AccordProgressLogTest extends TestBaseImpl
waitForRecovery.acquire();
long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(recoveryStartedAt.get() - coordinationStartedAt);
- Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 1000);
+ Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 100);
Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis <= 5000);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org