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

Improve setup and initialisation of LocalLog/LogSpec 

Patch by Alex Petrov; reviewed by Sam Tunnicliffe and marcuse for CASSANDRA-19271
This commit is contained in:
Alex Petrov 2023-11-17 17:13:30 +01:00 committed by Marcus Eriksson
parent 5ca972b145
commit 6b3958f1d8
24 changed files with 370 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Improve setup and initialisation of LocalLog/LogSpec (CASSANDRA-19271)
* Refactor structure of caching metrics and expose auth cache metrics via JMX (CASSANDRA-17062)
* Allow CQL client certificate authentication to work without sending an AUTHENTICATE request (CASSANDRA-18857)
* Extend nodetool tpstats and system_views.thread_pools with detailed pool parameters (CASSANDRA-19289)

9
src/java/org/apache/cassandra/schema/DistributedSchema.java
View File

@ -123,6 +123,10 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
{
keyspaceInstances.putAll(prev.keyspaceInstances);
// If there are keyspaces in schema, but none of them are initialised, we're in first boot. Initialise all.
if (!prev.isEmpty() && prev.keyspaceInstances.isEmpty())
prev = DistributedSchema.empty();
Keyspaces.KeyspacesDiff ksDiff = Keyspaces.diff(prev.getKeyspaces(), getKeyspaces());
SchemaChangeNotifier schemaChangeNotifier = Schema.instance.schemaChangeNotifier();
@ -148,8 +152,8 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
assert delta.before.name.equals(delta.after.name);
// drop tables and views
delta.views.dropped.forEach(v -> dropView(keyspace, v, true));
delta.tables.dropped.forEach(t -> dropTable(keyspace, t, true));
delta.views.dropped.forEach(v -> dropView(keyspace, v, loadSSTables));
delta.tables.dropped.forEach(t -> dropTable(keyspace, t, loadSSTables));
// add tables and views
delta.tables.created.forEach(t -> createTable(keyspace, t, loadSSTables));
@ -164,7 +168,6 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
keyspace.viewManager.reload(keyspaces.get(keyspace.getName()).get());
}
//schemaChangeNotifier.notifyKeyspaceAltered(delta);
SchemaDiagnostics.keyspaceAltered(Schema.instance, delta);
});

38
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.ExceptionCode;
import org.apache.cassandra.exceptions.StartupException;
import org.apache.cassandra.io.util.FileInputStreamPlus;
import org.apache.cassandra.io.util.FileOutputStreamPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
@ -47,10 +48,10 @@ import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.listeners.SchemaListener;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.Replication;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeVersion;
@ -150,22 +151,21 @@ public class ClusterMetadataService
ClusterMetadataService(PlacementProvider placementProvider,
Function<Processor, Processor> wrapProcessor,
Supplier<State> cmsStateSupplier,
LocalLog.LogSpec logSpec)
LocalLog.LogSpec logSpec) throws StartupException
{
this.placementProvider = placementProvider;
this.snapshots = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots();
Processor localProcessor;
LogStorage logStorage = LogStorage.SystemKeyspace;
if (CassandraRelevantProperties.TCM_USE_ATOMIC_LONG_PROCESSOR.getBoolean())
{
log = LocalLog.sync(logSpec);
log = logSpec.sync().createLog();
localProcessor = wrapProcessor.apply(new AtomicLongBackedProcessor(log, logSpec.isReset()));
fetchLogHandler = new FetchCMSLog.Handler((e, ignored) -> logStorage.getLogState(e));
fetchLogHandler = new FetchCMSLog.Handler((e, ignored) -> logSpec.storage().getLogState(e));
}
else
{
log = LocalLog.async(logSpec);
log = logSpec.async().createLog();
localProcessor = wrapProcessor.apply(new PaxosBackedProcessor(log));
fetchLogHandler = new FetchCMSLog.Handler();
}
@ -243,13 +243,23 @@ public class ClusterMetadataService
{
if (instance != null)
return;
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(Collections.singleton("DC1"));
emptyFromSystemTables.schema.initializeKeyspaceInstances(DistributedSchema.empty(), loadSSTables);
emptyFromSystemTables = emptyFromSystemTables.forceEpoch(Epoch.EMPTY);
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(emptyFromSystemTables)
.withStorage(new AtomicLongBackedProcessor.InMemoryStorage());
LocalLog log = LocalLog.sync(logSpec);
log.ready();
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(Collections.singleton("DC1"))
.forceEpoch(Epoch.EMPTY);
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(emptyFromSystemTables)
.loadSSTables(loadSSTables)
.withDefaultListeners(false)
.withListener(new SchemaListener(loadSSTables) {
@Override
public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
// we do not need a post-commit hook for tools
}
})
.sync()
.withStorage(new AtomicLongBackedProcessor.InMemoryStorage());
LocalLog log = logSpec.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
@ -260,6 +270,8 @@ public class ClusterMetadataService
null,
null,
new PeerLogFetcher(log));
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
ClusterMetadataService.setInstance(cms);
}

5
src/java/org/apache/cassandra/tcm/Period.java
View File

@ -60,7 +60,10 @@ public class Period
*/
public static long scanLogForPeriod(TableMetadata logTable, Epoch since)
{
long currentPeriod = ClusterMetadata.current().period;
long currentPeriod = Period.EMPTY;
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata != null)
currentPeriod = metadata.period;
PeriodFinder visitor = currentPeriod > Period.FIRST
? new ReversePeriodFinder(since, currentPeriod)
: new ForwardPeriodFinder(since);

52
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -47,7 +47,6 @@ import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.NewGossiper;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
@ -141,18 +140,16 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
public static void initializeAsNonCmsNode(Function<Processor, Processor> wrapProcessor) throws StartupException
{
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners();
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withStorage(LogStorage.SystemKeyspace)
.afterReplay(Startup::scrubDataDirectories)
.withDefaultListeners();
ClusterMetadataService.setInstance(new ClusterMetadataService(new UniformRangePlacement(),
wrapProcessor,
ClusterMetadataService::state,
logSpec));
ClusterMetadataService.instance().initRecentlySealedPeriodsIndex();
ClusterMetadataService.instance().log().replayPersisted();
ClusterMetadata replayed = ClusterMetadataService.instance().log().metadata();
scrubDataDirectories(replayed);
replayed.schema.initializeKeyspaceInstances(DistributedSchema.empty());
ClusterMetadataService.instance().log().ready();
ClusterMetadataService.instance().initRecentlySealedPeriodsIndex();
NodeId nodeId = ClusterMetadata.current().myNodeId();
UUID currentHostId = SystemKeyspace.getLocalHostId();
@ -179,6 +176,10 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
}
}
public interface AfterReplay
{
void accept(ClusterMetadata t) throws StartupException;
}
/**
* Initialization for Discovery.
*
@ -243,15 +244,17 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
public static void initializeFromGossip(Function<Processor, Processor> wrapProcessor, Runnable initMessaging) throws StartupException
{
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(SystemKeyspace.allKnownDatacenters());
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(emptyFromSystemTables)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners();
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(emptyFromSystemTables)
.afterReplay(Startup::scrubDataDirectories)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners();
ClusterMetadataService.setInstance(new ClusterMetadataService(new UniformRangePlacement(),
wrapProcessor,
ClusterMetadataService::state,
logSpec));
scrubDataDirectories(emptyFromSystemTables);
emptyFromSystemTables.schema.initializeKeyspaceInstances(DistributedSchema.empty());
ClusterMetadataService.instance().log().ready();
initMessaging.run();
try
@ -280,8 +283,9 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
assert cmGossip.equals(initial) : cmGossip + " != " + initial;
}
public static void reinitializeWithClusterMetadata(String fileName, Function<Processor, Processor> wrapProcessor, Runnable initMessaging) throws IOException
public static void reinitializeWithClusterMetadata(String fileName, Function<Processor, Processor> wrapProcessor, Runnable initMessaging) throws IOException, StartupException
{
ClusterMetadata prev = ClusterMetadata.currentNullable();
// First set a minimal ClusterMetadata as some deserialization depends
// on ClusterMetadata.current() to access the partitioner
StubClusterMetadataService initial = StubClusterMetadataService.forClientTools();
@ -298,25 +302,21 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
if (!metadata.isCMSMember(FBUtilities.getBroadcastAddressAndPort()))
throw new IllegalStateException("When reinitializing with cluster metadata, we must be in the CMS");
// can use local dc here since we know local host in the cms:
ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(Collections.singleton(DatabaseDescriptor.getLocalDataCenter()));
metadata.schema.initializeKeyspaceInstances(DistributedSchema.empty());
metadata = metadata.forceEpoch(metadata.epoch.nextEpoch());
ClusterMetadataService.unsetInstance();
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(metadata)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners()
.isReset(true)
.withReadyNotification(LocalLog.LogSpec.WhenReady.NONE);
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withPreviousState(prev)
.withInitialState(metadata)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners()
.isReset(true);
ClusterMetadataService.setInstance(new ClusterMetadataService(new UniformRangePlacement(),
wrapProcessor,
ClusterMetadataService::state,
logSpec));
// When re-intializing from a loaded metadata instance we need to fire notifications using the delta between an
// empty metadata and the loaded one. So we configure the LogSpec not to do any notifications and handle it
// explicitly here.
ClusterMetadataService.instance().log().notifyListeners(emptyFromSystemTables);
ClusterMetadataService.instance().log().ready();
initMessaging.run();
ClusterMetadataService.instance().forceSnapshot(metadata.forceEpoch(metadata.nextEpoch()));

8
src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java
View File

@ -67,12 +67,16 @@ public class StubClusterMetadataService extends ClusterMetadataService
{
super(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
LocalLog.sync(new LocalLog.LogSpec().withInitialState(initial)),
LocalLog.logSpec()
.loadSSTables(false)
.sync()
.withInitialState(initial)
.createLog(),
new StubProcessor(),
Commit.Replicator.NO_OP,
false);
this.metadata = initial;
this.log().ready();
this.log().readyUnchecked();
}
@Override

10
src/java/org/apache/cassandra/tcm/listeners/SchemaListener.java
View File

@ -29,17 +29,23 @@ import org.apache.cassandra.tcm.ClusterMetadata;
public class SchemaListener implements ChangeListener
{
private final boolean loadSSTables;
public SchemaListener(boolean loadSSTables)
{
this.loadSSTables = loadSSTables;
}
@Override
public void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
notifyInternal(prev, next, fromSnapshot, true);
notifyInternal(prev, next, fromSnapshot, loadSSTables);
}
protected void notifyInternal(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot, boolean loadSSTables)
{
if (!fromSnapshot && next.schema.lastModified().equals(prev.schema.lastModified()))
return;
next.schema.initializeKeyspaceInstances(prev.schema, loadSSTables);
}

231
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -42,10 +42,12 @@ import org.apache.cassandra.concurrent.Interruptible;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.exceptions.StartupException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.Startup;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.tcm.listeners.ClientNotificationListener;
@ -96,25 +98,57 @@ public abstract class LocalLog implements Closeable
// notification to them all.
private final AtomicBoolean replayComplete = new AtomicBoolean();
public static LogSpec logSpec()
{
return new LogSpec();
}
public static class LogSpec
{
public enum WhenReady { NONE, PRE_COMMIT_ONLY, POST_COMMIT_ONLY, ALL};
private ClusterMetadata initial = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
private ClusterMetadata initial;
private ClusterMetadata prev;
private Startup.AfterReplay afterReplay = (metadata) -> {};
private LogStorage storage = LogStorage.None;
private boolean async = true;
private boolean defaultListeners = false;
private boolean reset = false;
private WhenReady whenReady = WhenReady.POST_COMMIT_ONLY;
private boolean isReset = false;
private boolean loadSSTables = true;
private final Set<LogListener> listeners = new HashSet<>();
private final Set<ChangeListener> changeListeners = new HashSet<>();
private final Set<ChangeListener.Async> asyncChangeListeners = new HashSet<>();
private LogSpec()
{
}
/**
* create a sync log - only used for tests and tools
* @return
*/
public LogSpec sync()
{
this.async = false;
return this;
}
public LogSpec async()
{
this.async = true;
return this;
}
public LogSpec withDefaultListeners()
{
return withDefaultListeners(true);
}
public LogSpec loadSSTables(boolean loadSSTables)
{
this.loadSSTables = loadSSTables;
return this;
}
public LogSpec withDefaultListeners(boolean withDefaultListeners)
{
if (withDefaultListeners &&
@ -148,13 +182,18 @@ public abstract class LocalLog implements Closeable
public LogSpec isReset(boolean isReset)
{
reset = isReset;
this.isReset = isReset;
return this;
}
public boolean isReset()
{
return reset;
return this.isReset;
}
public LogStorage storage()
{
return storage;
}
public LogSpec withStorage(LogStorage storage)
@ -163,17 +202,31 @@ public abstract class LocalLog implements Closeable
return this;
}
public LogSpec afterReplay(Startup.AfterReplay afterReplay)
{
this.afterReplay = afterReplay;
return this;
}
public LogSpec withInitialState(ClusterMetadata initial)
{
this.initial = initial;
return this;
}
public LogSpec withReadyNotification(WhenReady whenReady)
public LogSpec withPreviousState(ClusterMetadata prev)
{
this.whenReady = whenReady;
this.prev = prev;
return this;
}
public final LocalLog createLog()
{
if (async)
return new Async(this);
else
return new Sync(this);
}
}
/**
@ -196,21 +249,27 @@ public abstract class LocalLog implements Closeable
return e1.epoch.compareTo(e2.epoch);
});
protected final LogStorage persistence;
protected final LogStorage storage;
protected final Set<LogListener> listeners;
protected final Set<ChangeListener> changeListeners;
protected final Set<ChangeListener.Async> asyncChangeListeners;
private final LogSpec spec;
protected final LogSpec spec;
private LocalLog(LogSpec spec)
private LocalLog(LogSpec logSpec)
{
assert spec.initial.epoch.is(EMPTY) || spec.initial.epoch.is(Epoch.UPGRADE_STARTUP) || spec.reset;
committed = new AtomicReference<>(spec.initial);
this.persistence = spec.storage;
this.spec = logSpec;
if (spec.initial == null)
spec.initial = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
if (spec.prev == null)
spec.prev = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
assert spec.initial.epoch.is(EMPTY) || spec.initial.epoch.is(Epoch.UPGRADE_STARTUP) || spec.isReset :
String.format(String.format("Should start with empty epoch, unless we're in upgrade or reset mode: %s (isReset: %s)", spec.initial, spec.isReset));
this.committed = new AtomicReference<>(logSpec.initial);
this.storage = logSpec.storage;
listeners = Sets.newConcurrentHashSet();
changeListeners = Sets.newConcurrentHashSet();
asyncChangeListeners = Sets.newConcurrentHashSet();
this.spec = spec;
}
public void bootstrap(InetAddressAndPort addr)
@ -250,34 +309,6 @@ public abstract class LocalLog implements Closeable
return pending.size();
}
public static LocalLog sync(LogSpec spec)
{
return new Sync(spec);
}
public static LocalLog async(LogSpec spec)
{
return new Async(spec);
}
@VisibleForTesting
public static LocalLog asyncForTests()
{
LogSpec logSpec = new LogSpec();
LocalLog log = new Async(logSpec);
log.ready();
return log;
}
@VisibleForTesting
public static LocalLog asyncForTests(ClusterMetadata initial)
{
LogSpec logSpec = new LogSpec().withInitialState(initial);
LocalLog log = new Async(logSpec);
log.ready();
return log;
}
public boolean hasGaps()
{
Epoch start = committed.get().epoch;
@ -305,7 +336,7 @@ public abstract class LocalLog implements Closeable
public Replication getCommittedEntries(Epoch since)
{
return persistence.getReplication(since);
return storage.getReplication(since);
}
public ClusterMetadata waitForHighestConsecutive()
@ -448,7 +479,7 @@ public abstract class LocalLog implements Closeable
next.epoch, pendingEntry.transform, prev.epoch);
if (replayComplete.get())
persistence.append(transformed.success().metadata.period, pendingEntry.maybeUnwrapExecuted());
storage.append(transformed.success().metadata.period, pendingEntry.maybeUnwrapExecuted());
notifyPreCommit(prev, next, isSnapshot);
@ -505,13 +536,13 @@ public abstract class LocalLog implements Closeable
/**
* Replays items that were persisted during previous starts. Replayed items _will not_ be persisted again.
*/
public Epoch replayPersisted()
private ClusterMetadata replayPersisted()
{
if (replayComplete.get())
throw new IllegalStateException("Can only replay persisted once.");
LogState logState = persistence.getLogState(metadata().epoch);
LogState logState = storage.getLogState(metadata().epoch);
append(logState);
return waitForHighestConsecutive().epoch;
return waitForHighestConsecutive();
}
private void maybeNotifyListeners(Entry entry, Transformation.Result result)
@ -538,11 +569,12 @@ public abstract class LocalLog implements Closeable
this.changeListeners.remove(listener);
}
public void notifyListeners(ClusterMetadata emptyFromSystemTables)
public void notifyListeners(ClusterMetadata prev)
{
ClusterMetadata metadata = ClusterMetadata.current();
notifyPreCommit(emptyFromSystemTables, metadata, true);
notifyPostCommit(emptyFromSystemTables, metadata, true);
ClusterMetadata metadata = committed.get();
logger.info("Notifying listeners, prev epoch = {}, current epoch = {}", prev.epoch, metadata.epoch);
notifyPreCommit(prev, metadata, true);
notifyPostCommit(prev, metadata, true);
}
private void notifyPreCommit(ClusterMetadata before, ClusterMetadata after, boolean fromSnapshot)
@ -561,15 +593,59 @@ public abstract class LocalLog implements Closeable
ScheduledExecutors.optionalTasks.submit(() -> listener.notifyPostCommit(before, after, fromSnapshot));
}
/**
* Essentially same as `ready` but throws an unchecked exception
*/
@VisibleForTesting
public final ClusterMetadata readyUnchecked()
{
try
{
return ready();
}
catch (StartupException e)
{
throw new RuntimeException(e);
}
}
public ClusterMetadata ready() throws StartupException
{
ClusterMetadata metadata = replayPersisted();
spec.afterReplay.accept(metadata);
logger.info("Marking LocalLog ready at epoch {}", metadata.epoch);
if (!replayComplete.compareAndSet(false, true))
throw new IllegalStateException("Log is already fully initialised");
logger.debug("Marking LocalLog ready at epoch {}", committed.get().epoch);
if (spec.defaultListeners)
{
logger.info("Adding default listeners to LocalLog");
addListeners();
}
else
{
logger.info("Adding specified listeners to LocalLog");
spec.listeners.forEach(this::addListener);
spec.changeListeners.forEach(this::addListener);
spec.asyncChangeListeners.forEach(this::addListener);
}
logger.info("Notifying all registered listeners of both pre and post commit event");
notifyListeners(spec.prev);
return metadata;
}
private static class Async extends LocalLog
{
private final AsyncRunnable runnable;
private final Interruptible executor;
private Async(LogSpec spec)
private Async(LogSpec logSpec)
{
super(spec);
super(logSpec);
this.runnable = new AsyncRunnable();
this.executor = ExecutorFactory.Global.executorFactory().infiniteLoop("GlobalLogFollower", runnable, SAFE, NON_DAEMON, UNSYNCHRONIZED);
}
@ -743,9 +819,9 @@ public abstract class LocalLog implements Closeable
private static class Sync extends LocalLog
{
private Sync(LogSpec spec)
private Sync(LogSpec logSpec)
{
super(spec);
super(logSpec);
}
void runOnce(DurationSpec durationSpec)
@ -780,7 +856,7 @@ public abstract class LocalLog implements Closeable
addListener(snapshotListener());
addListener(new InitializationListener());
addListener(new SchemaListener());
addListener(new SchemaListener(spec.loadSSTables));
addListener(new LegacyStateListener());
addListener(new PlacementsChangeListener());
addListener(new MetadataSnapshotListener());
@ -788,45 +864,6 @@ public abstract class LocalLog implements Closeable
addListener(new UpgradeMigrationListener());
}
public void ready()
{
if (!replayComplete.compareAndSet(false, true))
throw new IllegalStateException("Log is already fully initialised");
logger.debug("Marking LocalLog ready at epoch {}", committed.get().epoch);
if (spec.defaultListeners)
{
logger.debug("Adding default listeners to LocalLog");
addListeners();
}
else
{
logger.debug("Adding specified listeners to LocalLog");
spec.listeners.forEach(this::addListener);
spec.changeListeners.forEach(this::addListener);
spec.asyncChangeListeners.forEach(this::addListener);
}
switch (spec.whenReady)
{
case ALL:
logger.debug("Notifying all registered listeners of both pre and post commit event");
notifyListeners(spec.initial);
break;
case PRE_COMMIT_ONLY:
logger.debug("Notifying all registered listeners of pre-commit event only");
notifyPreCommit(spec.initial, committed.get(), true);
break;
case POST_COMMIT_ONLY:
logger.debug("Notifying all registered listeners of post-commit event only");
notifyPostCommit(spec.initial, committed.get(), true);
break;
case NONE:
logger.debug("Not notifying registered listeners of pre or post commit events");
break;
}
}
private LogListener snapshotListener()
{
return (entry, metadata) -> {

1
src/java/org/apache/cassandra/tcm/log/LogStorage.java
View File

@ -43,6 +43,7 @@ public interface LogStorage extends LogReader
{
return LogState.EMPTY;
}
public void truncate() {}
public Replication getReplication(Epoch since)
{
return Replication.EMPTY;

5
src/java/org/apache/cassandra/tcm/log/SystemKeyspaceStorage.java
View File

@ -101,6 +101,11 @@ public class SystemKeyspaceStorage implements LogStorage
return LogState.getLogState(since, snapshots.get(), this);
}
public void truncate()
{
Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(NAME).truncateBlockingWithoutSnapshot();
}
/**
* Uses the supplied period as a starting point to iterate through the log table
* collating log entries which follow the supplied epoch. It is assumed that the

3
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
View File

@ -110,8 +110,7 @@ public class JMXFeatureTest extends TestBaseImpl
Assert.assertThat(statusResult.getStderr(), is(blankOrNullString()));
Assert.assertThat(statusResult.getStdout(), containsString("DN 127.0.0.1"));
testInstance(instances, cluster.get(2));
ClusterUtils.start(instanceToStop, props -> {
});
ClusterUtils.start(instanceToStop, props -> {});
ClusterUtils.awaitRingState(otherInstance, instanceToStop, "Normal");
ClusterUtils.awaitRingStatus(otherInstance, instanceToStop, "Up");
statusResult = cluster.get(1).nodetoolResult("status");

9
test/distributed/org/apache/cassandra/distributed/test/log/CMSTestBase.java
View File

@ -96,9 +96,11 @@ public class CMSTestBase
this.rf = rf;
schemaProvider = Mockito.mock(SchemaProvider.class);
ClusterMetadata initial = new ClusterMetadata(partitioner);
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(initial).withDefaultListeners(addListeners);
log = LocalLog.sync(logSpec);
log.ready();
log = LocalLog.logSpec()
.sync()
.withInitialState(initial)
.withDefaultListeners(addListeners)
.createLog();
service = new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
@ -108,6 +110,7 @@ public class CMSTestBase
true);
ClusterMetadataService.setInstance(service);
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
service.commit(new Initialize(ClusterMetadata.current()) {
public Result execute(ClusterMetadata prev)

64
test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
View File

@ -22,8 +22,8 @@ import java.net.UnknownHostException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Random;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
@ -46,9 +46,9 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.Schema;
@ -63,6 +63,8 @@ import org.apache.cassandra.tcm.MetadataSnapshots;
import org.apache.cassandra.tcm.Period;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.log.Replication;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
@ -73,9 +75,9 @@ import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
import org.apache.cassandra.tcm.sequences.BootstrapAndJoin;
import org.apache.cassandra.tcm.sequences.BootstrapAndReplace;
import org.apache.cassandra.tcm.sequences.Move;
import org.apache.cassandra.tcm.sequences.LeaveStreams;
import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
import org.apache.cassandra.tcm.sequences.Move;
import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave;
import org.apache.cassandra.tcm.transformations.AlterSchema;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
@ -123,13 +125,15 @@ public class ClusterMetadataTestHelper
public static ClusterMetadataService instanceForTest()
{
ClusterMetadata current = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
LocalLog log = LocalLog.asyncForTests();
LocalLog log = LocalLog.logSpec()
.createLog();
ResettableClusterMetadataService service = new ResettableClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
true);
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
service.commit(new Initialize(current));
QueryProcessor.registerStatementInvalidatingListener();
@ -137,26 +141,6 @@ public class ClusterMetadataTestHelper
return service;
}
/**
* Create a pre-configured CMS which supports mark & reset for use in tests. This version dose not perform initial
* CMS setup, neither bootstrapping the log nor applying an Initialize transformation. It assumes that the supplied
* ClusterMetadata instance is in the state required by the specific caller.
* @return a resettable CMS instance, to be used in a call to ClusterMetadataService::setInstance
*/
public static ClusterMetadataService instanceForTest(ClusterMetadata current)
{
LocalLog log = LocalLog.asyncForTests();
ResettableClusterMetadataService service = new ResettableClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
true);
QueryProcessor.registerStatementInvalidatingListener();
service.mark();
return service;
}
public static ClusterMetadata minimalForTesting(IPartitioner partitioner)
{
return new ClusterMetadata(Epoch.EMPTY,
@ -186,13 +170,13 @@ public class ClusterMetadataTestHelper
null,
ImmutableMap.of());
}
public static void forceCurrentPeriodTo(long period)
{
ClusterMetadataService.unsetInstance();
ClusterMetadataService.setInstance(instanceForTest());
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata == null)
metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
metadata = new ClusterMetadata(metadata.epoch,
metadata = new ClusterMetadata(metadata.epoch.nextEpoch(),
period,
metadata.lastInPeriod,
metadata.partitioner,
@ -203,20 +187,23 @@ public class ClusterMetadataTestHelper
metadata.lockedRanges,
metadata.inProgressSequences,
metadata.extensions);
ClusterMetadataService.unsetInstance();
ClusterMetadataService.setInstance(instanceForTest(metadata));
ClusterMetadataService.instance().log().append(new LogState(metadata, Replication.of(Collections.emptyList())));
}
public static ClusterMetadataService syncInstanceForTest()
{
LocalLog log = LocalLog.sync(new LocalLog.LogSpec());
log.ready();
return new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
true);
LocalLog log = LocalLog.logSpec()
.sync()
.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
true);
log.readyUnchecked();
return cms;
}
public static void createKeyspace(String name, KeyspaceParams params)
@ -247,7 +234,6 @@ public class ClusterMetadataTestHelper
}
}
private static Set<InetAddressAndPort> leaving(ClusterMetadata metadata)
{
return metadata.directory.states.entrySet().stream()

18
test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java
View File

@ -642,9 +642,12 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
assert executor == null;
LogStorage logStorage = new AtomicLongBackedProcessor.InMemoryStorage();
ClusterMetadata initial = new ClusterMetadata(partitioner);
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(initial).withStorage(logStorage);
LocalLog log = LocalLog.sync(logSpec);
log.ready();
LocalLog log = LocalLog.logSpec()
.withInitialState(initial)
.sync()
.withStorage(logStorage)
.createLog();
// Replicator only replicates to the node under test, as there are no other nodes in reality
Commit.Replicator replicator = (result, source) -> {
realCluster.deliverMessage(realCluster.get(1).broadcastAddress(),
@ -661,7 +664,7 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
replicator,
true);
ClusterMetadataService.setInstance(service);
log.readyUnchecked();
log.bootstrap(cms.addr());
service.commit(new Initialize(log.metadata()));
service.commit(new Register(new NodeAddresses(cms.addr()), new Location(cms.dc(), cms.rack()), NodeVersion.CURRENT));
@ -720,8 +723,9 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
// We need to create a second node to be able to send and receive messages.
RealSimulatedNode driver = createNode();
LocalLog log = LocalLog.sync(new LocalLog.LogSpec());
log.ready();
LocalLog log = LocalLog.logSpec()
.sync()
.createLog();
ClusterMetadataService metadataService =
new ClusterMetadataService(new UniformRangePlacement(),
@ -754,6 +758,8 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
false);
ClusterMetadataService.setInstance(metadataService);
log.readyUnchecked();
driver.clean(TCM_REPLICATION);
driver.on(Verb.TCM_REPLICATION, new SimulatedAction<Replication, NoPayload>()
{

9
test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
View File

@ -94,16 +94,16 @@ public class DecommissionTest extends TestBaseImpl
}
@Test
public void testDecomDirectoryMinMaxVersions() throws IOException
{
try (Cluster cluster = builder().withNodes(3)
public void testDecomDirectoryMinMaxVersions() throws IOException {
try (Cluster cluster = builder()
.withConfig(cfg -> cfg.with(GOSSIP))
.withNodes(3)
.start())
{
cluster.get(3).nodetoolResult("decommission", "--force").asserts().success();
cluster.get(1).runOnInstance(() -> {
ClusterMetadata metadata = ClusterMetadata.current();
ClusterMetadataService.instance().commit(new Startup(metadata.myNodeId(),
metadata.directory.getNodeAddresses(metadata.myNodeId()),
new NodeVersion(new CassandraVersion("6.0.0"),
@ -133,6 +133,7 @@ public class DecommissionTest extends TestBaseImpl
@Test
public void testMixedVersionBlockDecom() throws IOException {
try (Cluster cluster = builder().withNodes(3)
.withConfig(config -> config.with(GOSSIP))
.start())
{
cluster.get(3).nodetoolResult("decommission", "--force").asserts().success();

41
test/unit/org/apache/cassandra/ServerTestUtils.java
View File

@ -24,8 +24,8 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -33,21 +33,20 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogManager;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.big.BigTableReader;
import org.apache.cassandra.io.sstable.indexsummary.IndexSummarySupport;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.locator.AbstractEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
@ -59,6 +58,7 @@ import org.apache.cassandra.tcm.MetadataSnapshots;
import org.apache.cassandra.tcm.Processor;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.PlacementProvider;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
@ -265,7 +265,7 @@ public final class ServerTestUtils
public static void initCMS()
{
// Effectively disable automatic snapshots using AtomicLongBackedProcessopr and LocaLLog.Sync interacts
// Effectively disable automatic snapshots using AtomicLongBackedProcessor and LocaLLog.Sync interacts
// badly with submitting SealPeriod transformations from the log listener. In this configuration, SealPeriod
// commits performed on NonPeriodicTasks threads end up actually performing the transformations as well as
// calling the pre and post commit listeners, which is not threadsafe. In a non-test setup the processing of
@ -276,9 +276,14 @@ public final class ServerTestUtils
IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
boolean addListeners = true;
ClusterMetadata initial = new ClusterMetadata(partitioner);
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(initial)
.withDefaultListeners(addListeners);
LocalLog log = LocalLog.async(logSpec);
if (!Keyspace.isInitialized())
Keyspace.setInitialized();
LocalLog log = LocalLog.logSpec()
.withInitialState(initial)
.withDefaultListeners(addListeners)
.createLog();
ResettableClusterMetadataService service = new ResettableClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
@ -287,10 +292,7 @@ public final class ServerTestUtils
true);
ClusterMetadataService.setInstance(service);
initial.schema.initializeKeyspaceInstances(DistributedSchema.empty());
if (!Keyspace.isInitialized())
Keyspace.setInitialized();
log.ready();
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
service.commit(new Initialize(ClusterMetadata.current()));
QueryProcessor.registerStatementInvalidatingListener();
@ -309,12 +311,13 @@ public final class ServerTestUtils
// |-- StorageService.instance.setPartitionerUnsafe(M3P) # test wants to use LongToken
// |-- ServerTestUtils.recreateCMS # recreates the CMS using the updated partitioner
ClusterMetadata initial = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
initial.schema.initializeKeyspaceInstances(DistributedSchema.empty());
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(initial)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners();
LocalLog log = LocalLog.async(logSpec);
log.ready();
LogStorage storage = LogStorage.SystemKeyspace;
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(initial)
.withStorage(storage)
.withDefaultListeners();
LocalLog log = logSpec.createLog();
ResettableClusterMetadataService cms = new ResettableClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
@ -323,6 +326,8 @@ public final class ServerTestUtils
true);
ClusterMetadataService.unsetInstance();
ClusterMetadataService.setInstance(cms);
((SystemKeyspaceStorage)LogStorage.SystemKeyspace).truncate();
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
cms.mark();
}

2
test/unit/org/apache/cassandra/hints/HintsUpgradeTest.java
View File

@ -26,6 +26,7 @@ import java.util.function.Consumer;
import com.google.common.collect.ImmutableMap;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
@ -47,6 +48,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@Ignore("TODO: TCM")
public class HintsUpgradeTest
{
static

67
test/unit/org/apache/cassandra/service/ClientWarningsTest.java
View File

@ -21,7 +21,6 @@ import java.util.Collection;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -34,6 +33,7 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.transport.Message;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.transport.SimpleClient;
@ -74,12 +74,12 @@ public class ClientWarningsTest extends CQLTester
{
client.connect(false);
QueryMessage query = new QueryMessage(createBatchStatement2(1), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
Message.Response resp = executeWithRetries(client,
new QueryMessage(createBatchStatement2(1), QueryOptions.DEFAULT));
assertNull(resp.getWarnings());
query = new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT);
resp = client.execute(query);
resp = executeWithRetries(client,
new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT));
assertEquals(1, resp.getWarnings().size());
}
}
@ -94,12 +94,14 @@ public class ClientWarningsTest extends CQLTester
{
client.connect(false);
QueryMessage query = new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold() / 2 + 1), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
Message.Response resp = executeWithRetries(client,
new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold() / 2 + 1), QueryOptions.DEFAULT));
assertEquals(1, resp.getWarnings().size());
query = new QueryMessage(createBatchStatement(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT);
resp = client.execute(query);
resp = executeWithRetries(client,
new QueryMessage(createBatchStatement(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT));
assertNull(resp.getWarnings());
}
}
@ -116,35 +118,50 @@ public class ClientWarningsTest extends CQLTester
for (int i = 0; i < iterations; i++)
{
QueryMessage query = new QueryMessage(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (1, %s, 1)",
KEYSPACE,
currentTable(),
i), QueryOptions.DEFAULT);
client.execute(query);
executeWithRetries(client,
new QueryMessage(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (1, %s, 1)",
KEYSPACE,
currentTable(),
i), QueryOptions.DEFAULT));
}
ColumnFamilyStore store = Keyspace.open(KEYSPACE).getColumnFamilyStore(currentTable());
Util.flush(store);
for (int i = 0; i < iterations; i++)
{
QueryMessage query = new QueryMessage(String.format("DELETE v FROM %s.%s WHERE pk = 1 AND ck = %s",
KEYSPACE,
currentTable(),
i), QueryOptions.DEFAULT);
client.execute(query);
executeWithRetries(client,
new QueryMessage(String.format("DELETE v FROM %s.%s WHERE pk = 1 AND ck = %s",
KEYSPACE,
currentTable(),
i), QueryOptions.DEFAULT));
}
Util.flush(store);
{
QueryMessage query = new QueryMessage(String.format("SELECT * FROM %s.%s WHERE pk = 1",
KEYSPACE,
currentTable()), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
Message.Response resp = executeWithRetries(client,
new QueryMessage(String.format("SELECT * FROM %s.%s WHERE pk = 1",
KEYSPACE,
currentTable()), QueryOptions.DEFAULT));
assertEquals(1, resp.getWarnings().size());
}
}
}
private static Message.Response executeWithRetries(SimpleClient client, QueryMessage query)
{
for (int i = 0; i < 10; i++)
{
try
{
return client.execute(query);
}
catch (RequestTimeoutException t)
{
logger.warn("Timed out. Retrying.");
}
}
throw new RuntimeException("Could not execute query after 10 tries");
}
@Test
public void testLargeBatchWithProtoV2() throws Exception
{
@ -154,8 +171,8 @@ public class ClientWarningsTest extends CQLTester
{
client.connect(false);
QueryMessage query = new QueryMessage(createBatchStatement(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
Message.Response resp = executeWithRetries(client,
new QueryMessage(createBatchStatement(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT));
assertNull(resp.getWarnings());
}
}

3
test/unit/org/apache/cassandra/tcm/BootWithMetadataTest.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.tcm;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
@ -99,7 +98,7 @@ public class BootWithMetadataTest
epoch = doTest(Epoch.create(epoch.getEpoch() + 100), first);
}
private Epoch doTest(Epoch epoch, ClusterMetadata first) throws IOException
private Epoch doTest(Epoch epoch, ClusterMetadata first) throws Throwable
{
long seed = System.nanoTime();
logger.info("STARTING TEST FROM EPOCH {}, SEED: {}", epoch, seed);

7
test/unit/org/apache/cassandra/tcm/DiscoverySimulationTest.java
View File

@ -60,8 +60,10 @@ public class DiscoverySimulationTest
public static void setup()
{
ClusterMetadata initial = new ClusterMetadata(Murmur3Partitioner.instance);
LocalLog log = LocalLog.sync(new LocalLog.LogSpec().withInitialState(initial));
log.ready();
LocalLog log = LocalLog.logSpec()
.sync()
.withInitialState(initial)
.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
@ -70,6 +72,7 @@ public class DiscoverySimulationTest
Commit.Replicator.NO_OP,
false);
ClusterMetadataService.setInstance(cms);
log.readyUnchecked();
}
@Test

16
test/unit/org/apache/cassandra/tcm/LogStateTest.java
View File

@ -29,10 +29,10 @@ import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.extensions.ExtensionValue;
import org.apache.cassandra.tcm.listeners.MetadataSnapshotListener;
import org.apache.cassandra.tcm.listeners.SchemaListener;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
@ -54,10 +54,13 @@ public class LogStateTest
LogStorage logStorage = LogStorage.SystemKeyspace;
MetadataSnapshots snapshots = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots();
ClusterMetadata initial = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(initial)
.withStorage(logStorage)
.withLogListener(new MetadataSnapshotListener());
LocalLog log = LocalLog.sync(logSpec);
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.sync()
.withInitialState(initial)
.withStorage(logStorage)
.withLogListener(new MetadataSnapshotListener())
.withListener(new SchemaListener(true));
LocalLog log = logSpec.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
snapshots,
log,
@ -66,8 +69,7 @@ public class LogStateTest
false);
ClusterMetadataService.unsetInstance();
ClusterMetadataService.setInstance(cms);
initial.schema.initializeKeyspaceInstances(DistributedSchema.empty());
log.ready();
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort());
}

25
test/unit/org/apache/cassandra/tcm/log/LocalLogTest.java
View File

@ -63,8 +63,11 @@ public class LocalLogTest
@Test
public void appendToFillGapWithConsecutiveBufferedEntries()
{
LocalLog log = LocalLog.sync(new LocalLog.LogSpec().withInitialState(cm()));
log.ready();
LocalLog log = LocalLog.logSpec()
.sync()
.withInitialState(cm())
.createLog();
log.readyUnchecked();
Epoch start = log.metadata().epoch;
assertEquals(EMPTY, start);
@ -87,10 +90,13 @@ public class LocalLogTest
@Test
public void sealPeriodForceSnapshotCollisionWithGap()
{
LocalLog log = LocalLog.sync(new LocalLog.LogSpec().withInitialState(cm()));
log.ready();
LocalLog log = LocalLog.logSpec()
.sync()
.withInitialState(cm())
.createLog();
log.readyUnchecked();
List<Entry> entries =new ArrayList<>();
List<Entry> entries = new ArrayList<>();
for (int i = 1; i <= 9; i++)
entries.add(entry(i));
entries.add(new Entry(Entry.Id.NONE,
@ -112,8 +118,11 @@ public class LocalLogTest
@Test
public void multipleSnapshotEntries()
{
LocalLog log = LocalLog.sync(new LocalLog.LogSpec().withInitialState(cm()));
log.ready();
LocalLog log = LocalLog.logSpec()
.sync()
.withInitialState(cm())
.createLog();
log.readyUnchecked();
List<Entry> entries =new ArrayList<>();
for (int i = 1; i <= 9; i++)
@ -165,7 +174,7 @@ public class LocalLogTest
CountDownLatch finish = CountDownLatch.newCountDownLatch(threads);
CountDownLatch finishReaders = CountDownLatch.newCountDownLatch(threads);
ExecutorPlus executor = executorFactory().configurePooled("APPENDER", threads * 2).build();
LocalLog log = LocalLog.asyncForTests(cm());
LocalLog log = LocalLog.logSpec().withInitialState(cm()).createLog();
List<Entry> committed = new CopyOnWriteArrayList<>(); // doesn't need to be concurrent, since log is single-threaded
log.addListener((e, m) -> committed.add(e));

9
test/unit/org/apache/cassandra/tcm/log/LogListenerNotificationTest.java
View File

@ -94,10 +94,13 @@ public class LogListenerNotificationTest
counter++;
}
};
LocalLog.LogSpec logSpec = new LocalLog.LogSpec().withInitialState(cm()).withLogListener(listener);
LocalLog log = LocalLog.sync(logSpec);
LocalLog log = LocalLog.logSpec()
.sync()
.withInitialState(cm())
.withLogListener(listener)
.createLog();
log.readyUnchecked();
log.append(new Entry(Entry.Id.NONE, Epoch.FIRST, PreInitialize.forTesting()));
log.ready();
log.append(input);
}

3
tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java
View File

@ -64,7 +64,6 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.schema.Types;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.JavaDriverUtils;
@ -620,7 +619,6 @@ public class StressCQLSSTableWriter implements Closeable
*/
public static ColumnFamilyStore createOfflineTable(CreateTableStatement.Raw schemaStatement, List<CreateTypeStatement.Raw> typeStatements, List<File> directoryList)
{
ClusterMetadata prev = ClusterMetadata.current();
String keyspace = schemaStatement.keyspace();
KeyspaceMetadata ksm = KeyspaceMetadata.create(keyspace, KeyspaceParams.simple(1));
@ -644,7 +642,6 @@ public class StressCQLSSTableWriter implements Closeable
Tables tables = Tables.of(tableMetadata);
KeyspaceMetadata updated = ksm.withSwapped(tables);
Schema.instance.submit((metadata) -> metadata.schema.getKeyspaces().withAddedOrUpdated(updated));
ClusterMetadata.current().schema.initializeKeyspaceInstances(prev.schema, false);
Keyspace.setInitialized();
Directories directories = new Directories(tableMetadata, directoryList.stream().map(f -> new Directories.DataDirectory(new org.apache.cassandra.io.util.File(f.toPath()))).collect(Collectors.toList()));