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

fixup

This commit is contained in:
Benedict Elliott Smith 2026-07-30 16:31:20 +01:00
parent ba683a02fd
commit 25f6d75be8
14 changed files with 94 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/modules/cassandra/pages/managing/configuration/configuration.adoc
View File

@ -153,7 +153,7 @@ configuration changes in the near future.
`@Replaces` is the annotation to be used when you make changes to any configuration parameters in `Config` class and `cassandra.yaml`, and you want to add backward
compatibility with previous Cassandra versions. `Converters` class enumerates the different methods used for backward compatibility.
`IDENTITY` is the one used for name change only. For more information about the other Converters, please, check the JavaDoc in the class.
For backward compatibility virtual table `LoadSettings` contains both the old and the new
For backward compatibility virtual table `Settings` contains both the old and the new
parameters with the old and the new value format. Only exception at the moment are the following three parameters: `key_cache_save_period`,
`row_cache_save_period` and `counter_cache_save_period` which appear only once with the new value format.
The old names and value format still can be used at least until the next major release. Deprecation warning is emitted on startup.

2
modules/accord

@ -1 +1 @@
Subproject commit 673cf2ff93a4b32a392c91feacda09bb48ac16e7
Subproject commit f4c8ae02f9393c5eb14720b5074bd45135ee5b00

12
src/java/org/apache/cassandra/config/AccordConfig.java
View File

@ -27,6 +27,7 @@ import accord.api.ProtocolModifiers.CoordinatorBacklogExecution;
import accord.api.ProtocolModifiers.FastExecution;
import accord.api.ProtocolModifiers.ReplicaExecution;
import accord.api.ProtocolModifiers.SendStableMessages;
import accord.api.ProtocolModifiers.UniqueTimestampOnConflict;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@ -34,7 +35,6 @@ import org.apache.cassandra.journal.Params;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.HLC_FIFO;
import static org.apache.cassandra.config.AccordConfig.QueueShardModel.THREAD_POOL_PER_SHARD;
import static org.apache.cassandra.config.AccordConfig.QueueSubmissionModel.SYNC;
import static org.apache.cassandra.config.AccordConfig.RangeIndexMode.in_memory;
@ -173,6 +173,13 @@ public class AccordConfig
BLENDED_PRIORITY_PHASE_FAIR,
}
public enum UniqueTimestampReservations
{
NONE,
SMALL_SHARED,
HISTOGRAM
}
public QueueShardModel queue_shard_model = THREAD_POOL_PER_SHARD;
public QueueSubmissionModel queue_submission_model = SYNC;
@ -334,6 +341,9 @@ public class AccordConfig
public Boolean send_minimal;
// note: simulator incompatible (for now)
public Boolean precise_micros;
public UniqueTimestampReservations unique_timestamp_reservations;
public UniqueTimestampOnConflict unique_timestamp_on_conflict;
public DurationSpec.IntMillisecondsBound unique_timestamp_reservation_range;
public boolean ephemeral_reads = true;
public boolean state_cache_listener_jfr_enabled = false;

10
src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java
View File

@ -44,6 +44,9 @@ import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
*/
public class ExceptionSerializer
{
// allow plenty of room for UTF8 encoding
private static final int MAX_MESSAGE_CHAR_LENGTH = Short.MAX_VALUE / 4;
public static class RemoteException extends RuntimeException
{
public final String originalClass;
@ -76,8 +79,11 @@ public class ExceptionSerializer
if (isFirstException) message = "Remote exception from host " + FBUtilities.getBroadcastAddressAndPort().toString() + (t.getLocalizedMessage() != null ? " - " + t.getLocalizedMessage() : "");
else message = t.getLocalizedMessage();
if (message.length() > Short.MAX_VALUE)
message = message.substring(0, Short.MAX_VALUE);
if (message == null)
return message;
if (message.length() > MAX_MESSAGE_CHAR_LENGTH)
message = message.substring(0, MAX_MESSAGE_CHAR_LENGTH);
return message;
}

9
src/java/org/apache/cassandra/service/accord/AccordDataStore.java
View File

@ -148,6 +148,7 @@ public class AccordDataStore implements DataStore
case SUCCESS:
callback.fetched(ranges);
syncResult.trySuccess(null);
// fall-through to ensure started
case START:
reportStarted();
break;
@ -157,7 +158,7 @@ public class AccordDataStore implements DataStore
break;
case ABORT:
case ERROR:
RuntimeException ex = new RuntimeException(String.format("Repair failed (%s):", event.getType(), event.getMessage()));
RuntimeException ex = new RuntimeException(String.format("Repair failed (%s): %s", event.getType(), event.getMessage()));
callback.fail(ranges, ex);
syncResult.tryFailure(ex);
break;
@ -173,7 +174,11 @@ public class AccordDataStore implements DataStore
node.commandStores().mapReduceConsume(new RestrictedStoreSelector(ranges, 0, Long.MAX_VALUE), new MapReduceConsumeCommandStores<Ranges, Timestamp>(ranges)
{
@Override public Timestamp reduce(Timestamp o1, Timestamp o2) { return Timestamp.max(o1, o2); }
@Override public void accept(Timestamp result, Throwable failure) { start.started(result); }
@Override public void accept(Timestamp result, Throwable failure)
{
if (failure != null) syncResult.tryFailure(failure);
else start.started(result);
}
@Override public TxnId primaryTxnId() { return null; }
@Override public String reason() { return "Compute MaxConflict to report for fetch"; }
@Override protected Timestamp applyInternal(SafeCommandStore safeStore) { return safeStore.commandStore().maxConflict(TxnId.NONE, ranges); }

9
src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
View File

@ -459,15 +459,6 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
{
super(sourceEpoch, syncId, ranges, partialTxn);
}
@Override
protected AsyncChain<Data> beginRead(SafeCommandStore safeStore, Timestamp executeAt, PartialTxn txn, Participants<?> execute)
{
AsyncChain<Data> result = super.beginRead(safeStore, executeAt, txn, execute);
// TODO (required): verify that streaming snapshots have all been created by now, so we won't stream any data that arrives after this
readStarted(safeStore);
return result;
}
}
@Override

68
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -63,6 +63,9 @@ import accord.local.CommandStores;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.ShardDistributor.EvenSplit;
import accord.local.TimeService;
import accord.local.UniqueTimeService;
import accord.local.UniqueTimeService.AtomicUniqueTime;
import accord.local.UniqueTimeService.AtomicUniqueTimeWithStaleReservation;
import accord.local.durability.DurabilityService;
import accord.local.durability.ShardDurability;
@ -93,7 +96,9 @@ import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.AccordConfig;
import org.apache.cassandra.config.AccordConfig.CatchupFallbackMode;
import org.apache.cassandra.config.AccordConfig.JournalConfig.ReplaySavePoint;
import org.apache.cassandra.config.AccordConfig.UniqueTimestampReservations;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -175,16 +180,20 @@ import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Write;
import static accord.primitives.TxnId.MediumPath.NoMediumPath;
import static accord.primitives.TxnId.MediumPath.TrackStable;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.concurrent.ExecutorFactory.SimulatorThreadTag.JOB;
import static org.apache.cassandra.concurrent.ExecutorFactory.SystemThreadTag.DAEMON;
import static org.apache.cassandra.config.AccordConfig.CatchupFallbackMode.EXIT;
import static org.apache.cassandra.config.AccordConfig.CatchupFallbackMode.REBOOTSTRAP;
import static org.apache.cassandra.config.AccordConfig.CatchupFallbackMode.REBOOTSTRAP_AND_CATCHUP;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.REBOOTSTRAP_INCOMPLETE;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.REBOOTSTRAP_RESET;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.RESET;
import static org.apache.cassandra.config.AccordConfig.UniqueTimestampReservations.HISTOGRAM;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccord;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordGlobalDurabilityCycle;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurabilityCycle;
@ -479,7 +488,7 @@ public class AccordService implements IAccordService, Shutdownable
this.node = new Node(localId,
messageSink,
topologyService,
time, new AtomicUniqueTimeWithStaleReservation(time),
time, uniqueTimeService(time),
() -> dataStore,
new KeyspaceSplitter(new EvenSplit<>(getAccord().commandStoreShardCount(), getPartitioner().accordSplitter())),
agent,
@ -529,6 +538,8 @@ public class AccordService implements IAccordService, Shutdownable
ProtocolModifiers.Configure.setSendStableMessages(config.send_stable);
if (config.send_minimal != null)
ProtocolModifiers.Configure.setSendMinimal(config.send_minimal);
if (config.unique_timestamp_on_conflict != null)
ProtocolModifiers.Configure.setUniqueTimestampOnConflict(config.unique_timestamp_on_conflict);
}
@Override
@ -718,8 +729,8 @@ public class AccordService implements IAccordService, Shutdownable
void catchup()
{
AccordConfig spec = DatabaseDescriptor.getAccord();
if (!spec.catchup_on_start)
AccordConfig config = DatabaseDescriptor.getAccord();
if (!config.catchup_on_start)
{
logger.info("Catchup disabled; continuing to startup");
return;
@ -731,22 +742,17 @@ public class AccordService implements IAccordService, Shutdownable
node.commandStores().forAllUnsafe(commandStore -> ((DefaultProgressLog)commandStore.unsafeProgressLog()).setMode(CATCH_UP));
try
{
AccordConfig.CatchupFallbackMode onError = spec.catchup_on_start_on_error;
AccordConfig.CatchupFallbackMode onTimeout = spec.catchup_on_start_on_timeout;
CatchupFallbackMode onError = config.catchup_on_start_on_error;
CatchupFallbackMode onTimeout = config.catchup_on_start_on_timeout;
long maxLatencyNanos = spec.catchup_on_start_fail_latency.toNanoseconds();
long maxLatencyNanos = config.catchup_on_start_fail_latency.toNanoseconds();
int attempts = 1;
while (true)
{
logger.info("Catchup with quorum...");
long start = nanoTime();
long failAt = start + maxLatencyNanos;
Future<Unsuccessful> f;
{
AsyncChain<Unsuccessful> submit = Catchup.catchup(node, failAt, NANOSECONDS);
f = toFuture(submit);
}
Future<Unsuccessful> f = toFuture(Catchup.catchup(node, failAt, NANOSECONDS));
f.awaitThrowUncheckedOnInterrupt();
Throwable failed = f.cause();
Unsuccessful result = f.getNow();
@ -769,7 +775,7 @@ public class AccordService implements IAccordService, Shutdownable
if (onError == REBOOTSTRAP)
return;
++attempts;
onError = onTimeout = spec.catchup_on_start_on_rebootstrap_fallback;
onError = onTimeout = rebootstrapFallbackMode(config);
continue;
}
}
@ -780,10 +786,10 @@ public class AccordService implements IAccordService, Shutdownable
if (result == null)
{
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
if (seconds <= config.catchup_on_start_success_latency.toSeconds())
return;
if (attempts < spec.catchup_on_start_max_attempts)
if (attempts < config.catchup_on_start_max_attempts)
{
logger.info("Catchup was slow, so we may behind again; retrying");
++attempts;
@ -810,7 +816,7 @@ public class AccordService implements IAccordService, Shutdownable
if (onTimeout == REBOOTSTRAP)
return;
++attempts;
onError = onTimeout = spec.catchup_on_start_on_rebootstrap_fallback;
onError = onTimeout = rebootstrapFallbackMode(config);
continue;
}
}
@ -826,6 +832,18 @@ public class AccordService implements IAccordService, Shutdownable
}
}
private static CatchupFallbackMode rebootstrapFallbackMode(AccordConfig config)
{
CatchupFallbackMode mode = config.catchup_on_start_on_rebootstrap_fallback;
if (mode == REBOOTSTRAP_AND_CATCHUP)
{
CatchupFallbackMode newMode = EXIT;
logger.warn("Converting {} to {} after first failed rebootstrap, to avoid infinite loop", mode, newMode);
mode = newMode;
}
return mode;
}
/**
* 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
@ -1469,4 +1487,22 @@ public class AccordService implements IAccordService, Shutdownable
{
return journal.configuration();
}
private static UniqueTimeService uniqueTimeService(TimeService time)
{
AccordConfig config = getAccord();
UniqueTimestampReservations reservations = config.unique_timestamp_reservations;
if (reservations == null)
reservations = HISTOGRAM;
switch (reservations)
{
default: throw UnhandledEnum.unknown(reservations);
case NONE: return new AtomicUniqueTime(time);
case SMALL_SHARED: return new AtomicUniqueTimeWithStaleReservation(time);
case HISTOGRAM:
long millis = config.unique_timestamp_reservation_range == null ? 50 : config.unique_timestamp_reservation_range.toMilliseconds();
return new UniqueTimeService.AtomicUniqueAutoStaleTimes(time, millis, MILLISECONDS);
}
}
}

5
src/java/org/apache/cassandra/service/accord/execution/SafeTask.java
View File

@ -467,13 +467,16 @@ public final class SafeTask<R> extends Task implements Cancellable, DebuggableTa
if (isSequencedByPriorityAtomic())
{
boolean isTxnIdSubset = context.isTxnIdSubsetOf(parent.context);
if (!isKeySubset)
{
Invariants.require(keysOrRanges.domain() == Key, "ATOMIC tasks over ranges must declare a subset of their parent's task keys() to avoid a range scan across which we would not impose sequencing");
// to avoid priority inversion deadlocks, if we are not a strict subset of the parent task we permit running with a single key ready
nonSync.alwaysReady = true;
}
boolean isTxnIdSubset = true;
for (TxnId txnId : context.txnIds())
isTxnIdSubset &= parent.refs.containsKey(txnId);
Invariants.require(isTxnIdSubset, "ATOMIC tasks must declare a subset of their parent's task txnIds() to avoid a priority inversion deadlock");
// TODO (required): we're appending to the fifo queue - does this maintain correct order?
setCacheQueuedFifoExclusive();

1
src/java/org/apache/cassandra/service/accord/execution/Task.java
View File

@ -262,7 +262,6 @@ public abstract class Task extends IntrusiveHeapNode implements Cancellable, Deb
Task next;
Task consequences;
// the queue position until run is invoked, at which point it is assigned a nanoTime() value
long position;
int info;

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

@ -273,13 +273,13 @@ public class CommandStoreSerializers
}
}
private short cast(long v)
private int cast(long v)
{
if (v < 0 || v >= (1 << RedundantStatus.Property.WAS_OWNED.ordinal()))
throw new IllegalStateException("Should not be serializing WAS_OWNED or NOT_OWNED statuses");
if ((v & ~0xFFFF) != 0) // retain this
throw new IllegalStateException("Cannot serialize RedundantStatus larger than 0xFFFF. Requires serialization changes.");
return (short)v;
return (int) v;
}
@Override
@ -301,7 +301,7 @@ public class CommandStoreSerializers
bounds[i] = CommandSerializers.txnId.deserialize(in);
int[] statuses = new int[count * 2];
for (int i = 0 ; i < statuses.length ; ++i)
statuses[i] = in.readShort();
statuses[i] = in.readShort() & 0xFFFF;
return new RedundantBefore.Bounds(range, startEpoch, endEpoch, bounds, statuses, staleUntilAtLeast);
}

6
test/distributed/org/apache/cassandra/distributed/test/accord/AccordJournalIntegrationTest.java
View File

@ -47,7 +47,7 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
{
try (WithProperties wp = new WithProperties().set(CassandraRelevantProperties.DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED, "true");
Cluster cluster = init(Cluster.build(1)
.withConfig(config -> config.set("accord.catchup_on_start", "DISABLED"))
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.withoutVNodes()
.start()))
{
@ -96,7 +96,7 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(1)
.withoutVNodes()
.withConfig(config -> config.set("accord.catchup_on_start", "DISABLED"))
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.start())
{
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
@ -128,7 +128,7 @@ public class AccordJournalIntegrationTest extends TestBaseImpl
.withoutVNodes()
.withConfig(c -> {
c.with(GOSSIP).with(NETWORK);
c.set("accord.catchup_on_start", "DISABLED");
c.set("accord.catchup_on_start", "false");
})
.start())
{

2
test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
View File

@ -98,7 +98,7 @@ public class AccordJournalReplayTest extends TestBaseImpl
.set("accord.retry_syncpoint", "1s*attempts")
.set("accord.retry_durability", "1s*attempts")
.set("accord.journal.replay_save_point", "NO")
.set("accord.catchup_on_start", "DISABLED")
.set("accord.catchup_on_start", "false")
.with(NETWORK, GOSSIP))
.start())
{

2
test/distributed/org/apache/cassandra/distributed/test/accord/journal/JournalAccessRouteIndexOnStartupRaceTest.java
View File

@ -66,7 +66,7 @@ public class JournalAccessRouteIndexOnStartupRaceTest extends TestBaseImpl
public void test() throws IOException
{
try (Cluster cluster = Cluster.build(1)
.withConfig(config -> config.set("accord.catchup_on_start", "DISABLED"))
.withConfig(config -> config.set("accord.catchup_on_start", "false"))
.withInstanceInitializer(BBHelper::install).start())
{
IInvokableInstance node = cluster.get(1);

4
test/distributed/org/apache/cassandra/fuzz/topology/AccordHardCatchupTest.java → test/distributed/org/apache/cassandra/fuzz/topology/AccordCatchupTest.java
View File

@ -46,7 +46,7 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
public class AccordHardCatchupTest extends FuzzTestBase
public class AccordCatchupTest extends FuzzTestBase
{
private static final int WRITES = 10;
private static final int POPULATION = 1000;
@ -117,7 +117,7 @@ public class AccordHardCatchupTest extends FuzzTestBase
writeAndValidate.run();
history.customThrowing(() -> {
cluster.get(2).config().set("accord.catchup_on_start", "HARD");
cluster.get(2).config().set("accord.catchup_on_start", "true");
cluster.get(2).startup();
cluster.get(2).logs().watchFor(".*Catchup.*");
cluster.get(1).logs().watchFor("/127.0.0.2:.* is now UP");