diff --git a/modules/accord b/modules/accord index b86550c154..a8916a18a0 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit b86550c154758e5dd223fa8101a2a83005b644a6 +Subproject commit a8916a18a012063f4da74aaccd39f0b828c99da0 diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java index e7159cda8d..0ab0447f6b 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java @@ -65,7 +65,7 @@ public final class DropKeyspaceStatement extends AlterSchemaStatement .collect(Collectors.joining(","))); List accordTables = keyspace.tables.stream() - .filter(TableMetadata::isAccordEnabled) + .filter(TableMetadata::requiresAccordSupport) .collect(Collectors.toList()); if (!accordTables.isEmpty()) throw ire("Cannot drop keyspace '%s' as it contains accord tables. (%s)", diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java index a008b45488..c9e6516c57 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java @@ -60,7 +60,7 @@ public final class DropTableStatement extends AlterSchemaStatement ? null : keyspace.getTableOrViewNullable(tableName); if (table == null // this can happen when ifExists=true... since its already been validated can skip - || !table.isAccordEnabled()) + || !table.requiresAccordSupport()) return super.commit(metadata); // Multi-Step Operation @@ -94,7 +94,7 @@ public final class DropTableStatement extends AlterSchemaStatement if (table.isView()) throw ire("Cannot use DROP TABLE on a materialized view. Please use DROP MATERIALIZED VIEW instead."); - if (table.isAccordEnabled() && table.params.pendingDrop) + if (table.requiresAccordSupport() && table.params.pendingDrop) throw ire("Table '%s.%s' is already being dropped", keyspaceName, tableName); Iterable views = keyspace.views.forTable(table.id); diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java index fc3e9044cf..9bf1595a7c 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java @@ -218,11 +218,11 @@ public class CassandraStreamReceiver implements StreamReceiver { CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion; checkNotNull(minVersion, "Unable to determine minimum cluster version"); - IAccordService accordService = AccordService.instance(); if (session.streamOperation().requiresBarrierTransaction() && cfs.metadata().requiresAccordSupport() && CassandraVersion.CASSANDRA_5_0.compareTo(minVersion) >= 0) { + IAccordService accordService = AccordService.instance(); Ranges accordRanges = AccordTopology.toAccordRanges(cfs.getTableId(), ranges); long startedAtNanos = nanoTime(); long timeoutNanos = DatabaseDescriptor.getAccordRangeSyncPointTimeoutNanos(); diff --git a/src/java/org/apache/cassandra/schema/DistributedSchema.java b/src/java/org/apache/cassandra/schema/DistributedSchema.java index b13cfc7db3..0eceb9b169 100644 --- a/src/java/org/apache/cassandra/schema/DistributedSchema.java +++ b/src/java/org/apache/cassandra/schema/DistributedSchema.java @@ -56,7 +56,7 @@ public class DistributedSchema implements MetadataValue { public static final Serializer serializer = new Serializer(); - public static final DistributedSchema empty() + public static DistributedSchema empty() { return new DistributedSchema(Keyspaces.none(), Epoch.EMPTY); } @@ -146,6 +146,11 @@ public class DistributedSchema implements MetadataValue return ks == null ? null : ks.tables.getNullable(cf); } + public boolean hasAccordKeyspaces() + { + return keyspaces.stream().anyMatch(ksm -> ksm.tables.stream().anyMatch(TableMetadata::requiresAccordSupport)); + } + public static DistributedSchema fromSystemTables(Keyspaces keyspaces, Set knownDatacenters) { if (!keyspaces.containsKeyspace(SchemaConstants.METADATA_KEYSPACE_NAME)) diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index be8bd058b9..d8acdbc266 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -824,7 +824,7 @@ public class CassandraDaemon { // Bootstrap with same address is an edge-case here, since we rely on HIBERNATE to prevent writes // toward the bootstrapping replacement, so there's no startup sequence involved. - if (StorageService.instance.isReplacingSameAddress() && StorageService.instance.isSurveyMode()) + if (StorageService.isReplacingSameAddress() && StorageService.instance.isSurveyMode()) return; // This node has not joined the ring (i.e. it was started with -Dcassandra.join_ring=false) diff --git a/src/java/org/apache/cassandra/service/Rebuild.java b/src/java/org/apache/cassandra/service/Rebuild.java index b6e4d61d1b..c7d40f08bf 100644 --- a/src/java/org/apache/cassandra/service/Rebuild.java +++ b/src/java/org/apache/cassandra/service/Rebuild.java @@ -158,10 +158,11 @@ public class Rebuild streamer.addKeyspaceToFetch(keyspace); } - StreamResultFuture resultFuture = streamer.fetchAsync(); - // wait for result - Future accordReady = AccordService.instance().epochReady(metadata.epoch); - Future ready = FutureCombiner.allOf(resultFuture, accordReady); + StreamResultFuture streamResult = streamer.fetchAsync(); + + Future accordReady = AccordService.instance().epochReadyFor(metadata); + Future ready = FutureCombiner.allOf(streamResult, accordReady); + // wait for result ready.get(); } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 87ce92f8f7..2c692932c3 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -844,18 +844,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } Gossiper.waitToSettle(); - NodeId self; - if (isReplacingSameAddress()) - { - self = ClusterMetadata.current().myNodeId(); - if (self == null) - throw new IllegalStateException("Tried to replace same address, but node does not seem to be registered"); - } - else - { - self = Register.maybeRegister(); - } - + NodeId self = Register.maybeRegister(); + AccordService.startup(self); RegistrationStatus.instance.onRegistration(); Startup.maybeExecuteStartupTransformation(self); diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java index 9f3563e367..37f8d74594 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java @@ -57,7 +57,6 @@ import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.Epoch; -import org.apache.cassandra.tcm.listeners.ChangeListener; import org.apache.cassandra.tcm.membership.NodeState; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Simulate; @@ -132,17 +131,6 @@ public class AccordConfigurationService extends AbstractConfigurationService collector = new AtomicReference<>(new PreInitStateCollector()); + + @Override + public void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + collector.get().notifyPreCommit(prev, next, fromSnapshot); + } + + @Override + public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + collector.get().notifyPostCommit(prev, next, fromSnapshot); + } + + @VisibleForTesting + public void resetForTesting(ClusterMetadata metadata) + { + PreInitStateCollector stateCollector = new PreInitStateCollector(); + stateCollector.items.add(metadata); + collector.set(stateCollector); + } + + /** + * Collects TCM events from startup util full Accord initialization to avoid races with TCM and creating gaps between + * epochs restored from journal and reported by TCM. + **/ + + static class PreInitStateCollector implements ChangeListener + { + private final List items = new ArrayList<>(4); + + @Override + public synchronized void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + items.add(next); + } + + public synchronized List getItems() + { + return new ArrayList<>(items); + } + } + } + private static final Logger logger = LoggerFactory.getLogger(AccordService.class); + private static final Future EPOCH_READY = ImmediateFuture.success(null); static { ProtocolModifiers.Toggles.setPermitLocalExecution(true); @@ -227,34 +281,42 @@ public class AccordService implements IAccordService, Shutdownable return instance().responseHandler(); } - public synchronized static void startup(NodeId tcmId) + @VisibleForTesting + public synchronized static AccordService startup(NodeId tcmId) { if (!DatabaseDescriptor.getAccordTransactionsEnabled()) { instance = NOOP_SERVICE; - return; + return null; } if (instance != null) - return; + return (AccordService) instance; AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId)); as.startup(); - if (StorageService.instance.isReplacingSameAddress()) - { - // when replacing another node but using the same ip the hostId will also match, this causes no TCM transactions - // to be committed... - // In order to bootup correctly, need to pull in the current epoch - ClusterMetadata current = ClusterMetadata.current(); - as.configService().listener.notifyPostCommit(current, current, false); - } instance = as; replayJournal(as); + as.finishInitialization(); + + as.configService.start(); + as.configService.unsafeMarkTruncated(); + as.fastPathCoordinator.start(); + + ClusterMetadataService.instance().log().addListener(as.fastPathCoordinator); + as.node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()), + Ints.checkedCast(getAccordShardDurabilityMaxSplits()), + Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS); + as.node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS); + as.state = State.STARTED; // Only enable durability scheduling _after_ we have fully replayed journal as.configService.registerListener(as.node.durability()); as.node.durability().start(); + + WatermarkCollector.fetchAndReportWatermarksAsync(as.configService); + return as; } @VisibleForTesting @@ -355,12 +417,6 @@ public class AccordService implements IAccordService, Shutdownable @Override public synchronized void startup() - { - unsafeStartupWithOverrides(null); - } - - @VisibleForTesting - public synchronized void unsafeStartupWithOverrides(@Nullable Journal.TopologyUpdate overrideNullTopologyUpdate) { if (state != State.INIT) return; @@ -370,7 +426,6 @@ public class AccordService implements IAccordService, Shutdownable ClusterMetadata metadata = ClusterMetadata.current(); configService.updateMapping(metadata); - long highestKnown = -1; List images = new ArrayList<>(); // Collect locally known topologies @@ -387,66 +442,55 @@ public class AccordService implements IAccordService, Shutdownable prev = next; } - if (prev == null) - prev = overrideNullTopologyUpdate; - // Instantiate latest topology from the log, if known if (prev != null) { node.commandStores().initializeTopologyUnsafe(prev); - highestKnown = prev.global.epoch(); } + // Replay local epochs + for (ImmutableTopoloyImage image : images) + configService.reportTopology(image.global); + } + + /** + * Startup is broken up in two phases: local and distributed startup. During local startup, we replay up to + * the latest epoch known to the node prior to restart. After that, we replay journal itself, and only after + * that we finish initializaiton and replay the rest of epochs. + */ + @VisibleForTesting + public void finishInitialization() + { + configService.updateMapping(ClusterMetadata.current()); + long highestKnown = -1; + if (configService.currentTopology() != null) + highestKnown = configService.currentEpoch(); try { TopologyRange remote = fetchTopologies(highestKnown + 1); - // Replay local epochs - for (ImmutableTopoloyImage image : images) - configService.reportTopology(image.global); - if (remote != null) remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE); // Subscribe to TCM events - ClusterMetadataService.instance().log().addListener(configService.listener); - - // We report current topology _after_ subscribing to TCM events since in a single-node cluster there - // will be no notification about the current epoch, since it's already reported. And in a multi-node cluster - // we do not want a race between addinga listener and reporting an epoch. - if (remote == null && images.isEmpty()) - configService.reportTopology(AccordTopology.createAccordTopology(metadata)); - - WatermarkCollector.fetchAndReportWatermarksAsync(configService()); - configService.unsafeMarkTruncated(); - - int attempt = 0; - int waitSeconds = 5; - long deadine = Clock.Global.nanoTime() + SECONDS.toNanos(60); - while (true) + ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener() { - Epoch await = Epoch.max(Epoch.create(configService.currentEpoch()), metadata.epoch); - try + @Override + public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) { - epochReady(await).get(waitSeconds, SECONDS); - break; - } - catch (TimeoutException e) - { - logger.warn("Epoch {} is not ready after waiting for {} seconds", metadata.epoch, (++attempt) * waitSeconds); - // In case there are any gaps, fetch unknown topologies. - metadata = ClusterMetadata.current(); - highestKnown = configService.currentEpoch(); - if (metadata.epoch.getEpoch() > highestKnown) - { - remote = fetchTopologies(highestKnown + 1); - if (remote != null) - remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE); - } + if (state != State.SHUTDOWN) + configService.maybeReportMetadata(next); } + }); - if (Clock.Global.nanoTime() > deadine) - throw new IllegalStateException(String.format("Could not initialize epoch %s. Config service state:\n%s", await, configService.getDebugStr())); + Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector), + "Listener should have been initialized with Accord pre-init state collector, but was " + prevListener.getClass()); + + MetadataChangeListener.PreInitStateCollector preinit = (MetadataChangeListener.PreInitStateCollector) prevListener; + for (ClusterMetadata item : preinit.getItems()) + { + if (item.epoch.getEpoch() > minEpoch()) + configService.maybeReportMetadata(item); } } catch (InterruptedException e) @@ -458,17 +502,7 @@ public class AccordService implements IAccordService, Shutdownable { throw new RuntimeException(e); } - - configService.start(); - fastPathCoordinator.start(); - ClusterMetadataService.instance().log().addListener(fastPathCoordinator); - node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()), - Ints.checkedCast(getAccordShardDurabilityMaxSplits()), - Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS); - node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS); - state = State.STARTED; } - /** * Queries peers to discover min epoch, and then fetches all topologies between min and current epochs */ @@ -938,9 +972,8 @@ public class AccordService implements IAccordService, Shutdownable return cmdTxnState.build(); } - @Nullable @Override - public Long minEpoch() + public long minEpoch() { return node.topology().minEpoch(); } @@ -973,6 +1006,15 @@ public class AccordService implements IAccordService, Shutdownable return promise; } + @Override + public Future epochReadyFor(ClusterMetadata metadata) + { + if (!metadata.schema.hasAccordKeyspaces()) + return EPOCH_READY; + + return epochReady(metadata.epoch); + } + @Override public void receive(Message message) { diff --git a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java index 78ec8c4589..1d5a21995a 100644 --- a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java +++ b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java @@ -18,6 +18,8 @@ package org.apache.cassandra.service.accord; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import com.google.common.collect.BiMap; @@ -55,6 +57,11 @@ class EndpointMapping implements AccordEndpointMapper return mapping.containsKey(id); } + public List nodes() + { + return new ArrayList<>(mapping.keySet()); + } + public Map removedNodes() { return removedNodes; diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 40c0e57a4c..1c5d624aab 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -28,6 +28,7 @@ import java.util.function.BiConsumer; import javax.annotation.Nonnull; import javax.annotation.Nullable; +import org.apache.cassandra.tcm.ClusterMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -126,6 +127,8 @@ public interface IAccordService */ Future epochReady(Epoch epoch); + Future epochReadyFor(ClusterMetadata epoch); + void receive(Message message); class AccordCompactionInfo @@ -173,8 +176,8 @@ public interface IAccordService Id nodeId(); List debugTxnBlockedGraph(TxnId txnId); - @Nullable - Long minEpoch(); + + long minEpoch(); void awaitDone(TableId id, long epoch); @@ -302,6 +305,12 @@ public interface IAccordService return BOOTSTRAP_SUCCESS; } + @Override + public Future epochReadyFor(ClusterMetadata epoch) + { + return BOOTSTRAP_SUCCESS; + } + @Override public void receive(Message message) {} @@ -329,11 +338,10 @@ public interface IAccordService return Collections.emptyList(); } - @Nullable @Override - public Long minEpoch() + public long minEpoch() { - return null; + return -1; } @Override @@ -498,6 +506,12 @@ public interface IAccordService return delegate.epochReady(epoch); } + @Override + public Future epochReadyFor(ClusterMetadata epoch) + { + return delegate.epochReadyFor(epoch); + } + @Override public void receive(Message message) { @@ -528,9 +542,8 @@ public interface IAccordService return delegate.debugTxnBlockedGraph(txnId); } - @Nullable @Override - public Long minEpoch() + public long minEpoch() { return delegate.minEpoch(); } diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java index 93b864d79b..9c39e5dbd2 100644 --- a/src/java/org/apache/cassandra/streaming/StreamPlan.java +++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java @@ -238,7 +238,7 @@ public class StreamPlan public static String[] nonAccordTablesForKeyspace(KeyspaceMetadata ksm) { String[] result = ksm.tables.stream() - .filter(tbl -> !tbl.isAccordEnabled()) + .filter(tbl -> !tbl.requiresAccordSupport()) .map(tbl -> tbl.name) .toArray(String[]::new); @@ -247,11 +247,11 @@ public class StreamPlan public static boolean hasNonAccordTables(KeyspaceMetadata ksm) { - return ksm.tables.stream().anyMatch(tbl -> !tbl.isAccordEnabled()); + return ksm.tables.stream().anyMatch(tbl -> !tbl.requiresAccordSupport()); } public static boolean hasAccordTables(KeyspaceMetadata ksm) { - return ksm.tables.stream().anyMatch(TableMetadata::isAccordEnabled); + return ksm.tables.stream().anyMatch(TableMetadata::requiresAccordSupport); } } diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index 64ccbf37e8..bb660b9845 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -53,7 +53,6 @@ import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.tcm.log.LocalLog; import org.apache.cassandra.tcm.log.LogStorage; import org.apache.cassandra.tcm.log.SystemKeyspaceStorage; @@ -76,7 +75,6 @@ import static org.apache.cassandra.tcm.compatibility.GossipHelper.emptyWithSchem import static org.apache.cassandra.tcm.compatibility.GossipHelper.fromEndpointStates; import static org.apache.cassandra.tcm.membership.NodeState.JOINED; import static org.apache.cassandra.tcm.membership.NodeState.LEFT; -import static org.apache.cassandra.tcm.membership.NodeState.REGISTERED; import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; /** @@ -423,9 +421,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; metadata = ClusterMetadata.current(); NodeState startingstate = metadata.directory.peerState(self); - if (startingstate != REGISTERED && startingstate != LEFT) - AccordService.startup(self); - switch (startingstate) { case REGISTERED: @@ -437,7 +432,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; // When Accord starts up it needs to check for any historic epochs that it needs to know about (in order // to handle pending transactions), in order to know what nodes to check with it needs to know what the // settled placement is (so it knows what peers to reach out to). - AccordService.startup(self); InProgressSequences.finishInProgressSequences(self, true); // potentially finish the MSO committed above metadata = ClusterMetadata.current(); diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index 425f4bdc24..fc093344b0 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java @@ -38,6 +38,7 @@ import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.google.common.collect.Sets; +import org.apache.cassandra.service.accord.AccordService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -507,7 +508,7 @@ public abstract class LocalLog implements Closeable } catch (Throwable t) { - logger.error(String.format("Caught an exception while processing entry %s. This can mean that this node is configured differently from CMS.", prev), t); + logger.error("Caught an exception while processing entry {}. This can mean that this node is configured differently from CMS.", prev, t); throw new StopProcessingException(t); } @@ -564,8 +565,8 @@ public abstract class LocalLog implements Closeable } else if (!pendingEntry.epoch.isAfter(metadata().epoch)) { - logger.debug(String.format("An already appended entry %s discovered in the pending buffer, ignoring. Max consecutive: %s", - pendingEntry.epoch, prev.epoch)); + logger.debug("An already appended entry {} discovered in the pending buffer, ignoring. Max consecutive: {}", + pendingEntry.epoch, prev.epoch); pending.remove(pendingEntry); } else @@ -937,6 +938,8 @@ public abstract class LocalLog implements Closeable addListener(new MetadataSnapshotListener()); addListener(new ClientNotificationListener()); addListener(new UpgradeMigrationListener()); + if (DatabaseDescriptor.getAccord().enabled) + addListener(AccordService.MetadataChangeListener.instance); } private LogListener snapshotListener() diff --git a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java index 0e32f18c25..341eff845b 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java +++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java @@ -25,7 +25,6 @@ import java.util.Set; import java.util.stream.StreamSupport; import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.Lists; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -363,14 +362,16 @@ public class BootstrapAndJoin extends MultiStepOperation StorageService.instance.repairPaxosForTopologyChange("bootstrap"); Future bootstrapStream = StorageService.instance.startBootstrap(metadata, beingReplaced, movements, strictMovements); - Future accordReady = AccordService.instance().epochReady(metadata.epoch); - Future ready = FutureCombiner.allOf(Lists.newArrayList(bootstrapStream, accordReady)); + Future accordReady = AccordService.instance().epochReadyFor(metadata); + Future ready = FutureCombiner.allOf(bootstrapStream, accordReady); + try { if (bootstrapTimeoutMillis > 0) ready.get(bootstrapTimeoutMillis, MILLISECONDS); else ready.get(); + StorageService.instance.markViewsAsBuilt(); StorageService.instance.clearOngoingBootstrap(); logger.info("Bootstrap completed for tokens {}", tokens); diff --git a/src/java/org/apache/cassandra/tcm/sequences/Move.java b/src/java/org/apache/cassandra/tcm/sequences/Move.java index 450889694e..b54b796749 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/Move.java +++ b/src/java/org/apache/cassandra/tcm/sequences/Move.java @@ -257,8 +257,10 @@ public class Move extends MultiStepOperation } StreamResultFuture streamResult = streamPlan.execute(); - Future accordReady = AccordService.instance().epochReady(metadata.epoch); - FutureCombiner.allOf(streamResult, accordReady).get(); + + Future accordReady = AccordService.instance().epochReadyFor(metadata); + Future ready = FutureCombiner.allOf(streamResult, accordReady); + ready.get(); StorageService.instance.repairPaxosForTopologyChange("move"); } catch (InterruptedException e) diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java index 2c960a0bd3..80c33816c7 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java +++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java @@ -58,7 +58,7 @@ public class PrepareDropAccordTable implements Transformation TableMetadata metadata = prev.schema.getKeyspaces().getTableOrViewNullable(tableRef.id); if (metadata == null) return new Rejected(ExceptionCode.INVALID, "Table " + tableRef + " is not known"); - if (!metadata.isAccordEnabled()) + if (!metadata.requiresAccordSupport()) return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is not an Accord table and should be dropped normally"); if (metadata.params.pendingDrop) return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is in the process of being dropped"); diff --git a/src/java/org/apache/cassandra/tcm/transformations/Register.java b/src/java/org/apache/cassandra/tcm/transformations/Register.java index 7cf45fab8b..c60244bab7 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/Register.java +++ b/src/java/org/apache/cassandra/tcm/transformations/Register.java @@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.exceptions.ExceptionCode.INVALID; +import static org.apache.cassandra.service.StorageService.isReplacingSameAddress; public class Register implements Transformation { @@ -86,6 +87,14 @@ public class Register implements Transformation public static NodeId maybeRegister() { + if (isReplacingSameAddress()) + { + NodeId self = ClusterMetadata.current().myNodeId(); + if (self == null) + throw new IllegalStateException("Tried to replace same address, but node does not seem to be registered"); + + return self; + } return register(false); } diff --git a/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java b/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java index e77a231da2..0ca992b642 100644 --- a/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java +++ b/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java @@ -27,13 +27,13 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +import ch.qos.logback.classic.LoggerContext; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import ch.qos.logback.classic.LoggerContext; import net.nicoulaj.compilecommand.annotations.Exclude; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.Config; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index c8e24ef092..70743f7b06 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -844,7 +844,6 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance JVMStabilityInspector.replaceKiller(new InstanceKiller(Instance.this::shutdown)); StorageService.instance.registerDaemon(CassandraDaemon.getInstanceForTesting()); - if (config.has(GOSSIP)) { try diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index a9fe78f642..51a551a19a 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -699,7 +699,7 @@ public class ClusterUtils return maxEpoch(cluster, up.toIntArray()); } - public static Epoch maxEpoch(ICluster cluster, int[] nodes) + public static Epoch maxEpoch(ICluster cluster, int... nodes) { Epoch max = null; for (int id : nodes) diff --git a/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java b/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java new file mode 100644 index 0000000000..ab23628fa3 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java @@ -0,0 +1,70 @@ +/* + * 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.distributed.test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.transformations.CustomTransformation; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_REQ; + +public class PaxosEpochSerializationTest extends TestBaseImpl +{ + private static final Logger logger = LoggerFactory.getLogger(PaxosEpochSerializationTest.class); + private static final String TABLE = "tbl"; + + @Test + public void epochBadDeserializationTest() throws Throwable + { + try (Cluster cluster = init(Cluster.build(3).withConfig(cfg -> cfg.with(Feature.NETWORK) + .with(Feature.GOSSIP)) + .withoutVNodes().start())) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk text primary key, v int)"); + cluster.get(1).runOnInstance(() -> { + // just execute transformations to get epoch bumped enough for the bug to occur + for (int i = 0; i < 75; i++) + ClusterMetadataService.instance().commit(CustomTransformation.make("x"+i)); + }); + // and bump the epoch in the tablemetadata: + cluster.schemaChange("ALTER TABLE " + KEYSPACE + '.' + TABLE + " WITH comment='abc'"); + + cluster.verbs(PAXOS_COMMIT_REQ).drop(); + try + { + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, v) VALUES ('xyzxyzxyzxyzxyzxyzxyzxyz', 1) IF NOT EXISTS", ConsistencyLevel.QUORUM); + Assert.fail("expected write timeout"); + } + catch (RuntimeException e) + { + // exception expected + } + + cluster.filters().reset(); + cluster.get(1).shutdown().get(); + cluster.get(1).startup(); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java index febffe5290..d18f3a6f06 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java @@ -91,7 +91,6 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.NodeVersion; -import org.apache.cassandra.tcm.transformations.CustomTransformation; import org.apache.cassandra.tcm.transformations.ForceSnapshot; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.CassandraVersion; @@ -295,38 +294,6 @@ public class PaxosRepairTest extends TestBaseImpl } } - @Test - public void epochBadDeserializationTest() throws Throwable - { - try (Cluster cluster = init(Cluster.build(3).withConfig(WITH_NETWORK).withoutVNodes().start())) - { - cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk text primary key, v int)"); - cluster.get(1).runOnInstance(() -> { - // just execute transformations to get epoch bumped enough for the bug to occur - for (int i = 0; i < 75; i++) - ClusterMetadataService.instance().commit(CustomTransformation.make("x"+i)); - }); - // and bump the epoch in the tablemetadata: - cluster.schemaChange("ALTER TABLE " + KEYSPACE + '.' + TABLE + " WITH comment='abc'"); - - cluster.verbs(PAXOS_COMMIT_REQ).drop(); - try - { - cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, v) VALUES ('xyzxyzxyzxyzxyzxyzxyzxyz', 1) IF NOT EXISTS", ConsistencyLevel.QUORUM); - Assert.fail("expected write timeout"); - } - catch (RuntimeException e) - { - // exception expected - } - - cluster.filters().reset(); - cluster.get(1).shutdown().get(); - cluster.get(1).startup(); - } - } - - @Test public void topologyChangePaxosTest() throws Throwable { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java index a5aae94516..7f9fef5a6a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java @@ -19,22 +19,10 @@ package org.apache.cassandra.distributed.test.accord; import java.util.Collection; -import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; - -import com.google.common.collect.Iterables; -import org.junit.After; -import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.Ignore; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import accord.local.Node; import accord.local.PreLoadContext; import accord.local.SafeCommand; @@ -49,14 +37,18 @@ import accord.primitives.Timestamp; import accord.primitives.TxnId; import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; +import com.google.common.collect.Iterables; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.api.IIsolatedExecutor; +import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.harry.checker.ModelChecker; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; @@ -68,14 +60,20 @@ import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.Clock; -import org.apache.cassandra.utils.concurrent.Future; -import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Ignore; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import static accord.local.LoadKeys.SYNC; import static accord.local.LoadKeysFor.READ_WRITE; import static java.lang.String.format; +import static org.apache.cassandra.distributed.test.accord.AccordTestBase.executeWithRetry; -public class AccordIncrementalRepairTest extends AccordTestBase +public class AccordIncrementalRepairTest extends TestBaseImpl { private static final Logger logger = LoggerFactory.getLogger(AccordIncrementalRepairTest.class); @@ -122,7 +120,6 @@ public class AccordIncrementalRepairTest extends AccordTestBase return AccordService.instance(); } - @Override protected Logger logger() { return logger; @@ -131,25 +128,41 @@ public class AccordIncrementalRepairTest extends AccordTestBase @BeforeClass public static void setupClass() throws Throwable { - setupCluster(opt -> opt.withConfig(conf -> conf.with(Feature.NETWORK, Feature.GOSSIP) - .set("accord.recover_txn", "1s") - .set("accord.retry_syncpoint", "1s*attempts") - .set("accord.retry_durability", "1s*attempts") - .set("accord.shard_durability_target_splits", 4) - ), 3); - for (IInvokableInstance instance : SHARED_CLUSTER) - instance.runOnInstance(() -> AccordService.unsafeSetNewAccordService(new BarrierRecordingService(AccordService.instance()))); -// setupCluster(opt -> opt, 3); + + } - @After - public void tearDown() + private static Cluster createCluster() throws Throwable { - for (IInvokableInstance instance : SHARED_CLUSTER) - instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start())); - SHARED_CLUSTER.filters().reset(); + Cluster cluster = AccordTestBase.createCluster(3, opt -> opt.withConfig(conf -> conf.with(Feature.NETWORK, Feature.GOSSIP) + .set("accord.recover_txn", "1s") + .set("accord.retry_syncpoint", "1s*attempts") + .set("accord.retry_durability", "1s*attempts") + .set("accord.shard_durability_target_splits", 4))); + for (IInvokableInstance instance : cluster) + instance.runOnInstance(() -> AccordService.unsafeSetNewAccordService(new BarrierRecordingService(AccordService.instance()))); + + return cluster; } + private static void withCluster(ModelChecker.ThrowingConsumer run) throws Throwable + { + try (Cluster cluster = createCluster()) + { + try + { + run.accept(cluster); + } + catch (Throwable t) + { + cluster.filters().reset(); + for (IInvokableInstance instance : cluster) + instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start())); + } + } + } + + private static void await(IInvokableInstance instance, IIsolatedExecutor.SerializableCallable check, long duration, TimeUnit unit) { instance.runOnInstance(() -> { @@ -184,27 +197,6 @@ public class AccordIncrementalRepairTest extends AccordTestBase await(instance, () -> !FailureDetector.instance.isAlive(endpoint), 1, TimeUnit.MINUTES); } - private static V getUninterruptibly(Future future, long timeout, TimeUnit units) - { - try - { - return future.get(timeout, units); - } - catch (InterruptedException e) - { - throw new UncheckedInterruptedException(e); - } - catch (ExecutionException | TimeoutException e) - { - throw new RuntimeException(e); - } - } - - private static V getUninterruptibly(Future future) - { - return getUninterruptibly(future, 1, TimeUnit.MINUTES); - } - private static TxnId awaitLocalApplyOnKey(TableMetadata metadata, int k) { return awaitLocalApplyOnKey(new TokenKey(metadata.id, metadata.partitioner.decorateKey(ByteBufferUtil.bytes(k)).getToken())); @@ -254,25 +246,31 @@ public class AccordIncrementalRepairTest extends AccordTestBase @Test public void txnRepairTest() throws Throwable { - SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName)); + withCluster(this::txnRepairTest); + } + + public void txnRepairTest(Cluster cluster) throws Throwable + { final String keyspace = KEYSPACE; - final String table = accordTableName; + final String table = "accord_table"; + final String qualifiedTable = String.format("%s.%s", keyspace, table); + cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", keyspace, table)); - SHARED_CLUSTER.filters().allVerbs().to(3).drop(); - awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3)); + cluster.filters().allVerbs().to(3).drop(); + awaitEndpointDown(cluster.get(1), cluster.get(3)); - executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" + - "INSERT INTO %s (k, v) VALUES (1, 1);\n" + - "COMMIT TRANSACTION", qualifiedAccordTableName)); + executeWithRetry(cluster, format("BEGIN TRANSACTION\n" + + "INSERT INTO %s (k, v) VALUES (1, 1);\n" + + "COMMIT TRANSACTION", qualifiedTable)); - SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { + cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table); awaitLocalApplyOnKey(metadata, 1); })); - SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); + cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); - SHARED_CLUSTER.get(1, 2).forEach(instance -> { + cluster.get(1, 2).forEach(instance -> { instance.runOnInstance(() -> { ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table); cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); @@ -283,23 +281,23 @@ public class AccordIncrementalRepairTest extends AccordTestBase }); }); }); - SHARED_CLUSTER.get(3).runOnInstance(() -> { + cluster.get(3).runOnInstance(() -> { ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table); cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); Assert.assertTrue(cfs.getLiveSSTables().isEmpty()); }); // heal partition and wait for node 1 to see node 3 again - for (IInvokableInstance instance : SHARED_CLUSTER) + for (IInvokableInstance instance : cluster) instance.runOnInstance(() -> { AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().stop()); Assert.assertFalse(barrierRecordingService().executedBarriers); }); - SHARED_CLUSTER.filters().reset(); - awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3)); - nodetool(SHARED_CLUSTER.get(1), "repair", KEYSPACE); + cluster.filters().reset(); + awaitEndpointUp(cluster.get(1), cluster.get(3)); + nodetool(cluster.get(1), "repair", keyspace); - SHARED_CLUSTER.get(1).runOnInstance(() -> { + cluster.get(1).runOnInstance(() -> { Assert.assertTrue(barrierRecordingService().executedBarriers); ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table); Assert.assertFalse(cfs.getLiveSSTables().isEmpty()); @@ -309,17 +307,17 @@ public class AccordIncrementalRepairTest extends AccordTestBase }); } - private void testSingleNodeWrite(TransactionalMode mode) + private void testSingleNodeWrite(Cluster cluster, TransactionalMode mode) { - SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='%s';", KEYSPACE, accordTableName, mode)); final String keyspace = KEYSPACE; - final String table = accordTableName; + final String table = "accord_table"; + cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='%s';", keyspace, table, mode)); - SHARED_CLUSTER.get(3).runOnInstance(() -> { + cluster.get(3).runOnInstance(() -> { QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s (k, v) VALUES (1, 2);", keyspace, table)); }); - SHARED_CLUSTER.get(3).runOnInstance(() -> { + cluster.get(3).runOnInstance(() -> { UntypedResultSet result = QueryProcessor.executeInternal(format("SELECT * FROM %s.%s WHERE k=1", keyspace, table)); Assert.assertFalse(result.isEmpty()); UntypedResultSet.Row row = Iterables.getOnlyElement(result); @@ -336,7 +334,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase Assert.assertFalse(sstable.isPendingRepair()); }); }); - SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { + cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { UntypedResultSet result = QueryProcessor.executeInternal(format("SELECT * FROM %s.%s WHERE k=1", keyspace, table)); Assert.assertTrue(result.isEmpty()); @@ -344,12 +342,12 @@ public class AccordIncrementalRepairTest extends AccordTestBase cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); Assert.assertTrue(cfs.getLiveSSTables().isEmpty()); })); - SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> { + cluster.forEach(instance -> instance.runOnInstance(() -> { barrierRecordingService().reset(); })); - nodetool(SHARED_CLUSTER.get(1), "repair", KEYSPACE); - SHARED_CLUSTER.get(1).runOnInstance(() -> { + nodetool(cluster.get(1), "repair", KEYSPACE); + cluster.get(1).runOnInstance(() -> { Assert.assertTrue(barrierRecordingService().executedBarriers); ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table); Assert.assertFalse(cfs.getLiveSSTables().isEmpty()); @@ -369,74 +367,90 @@ public class AccordIncrementalRepairTest extends AccordTestBase * a failed write at txn mode unsafe should be made visible by repair */ @Test - public void unsafeRepairTest() + public void unsafeRepairTest() throws Throwable { - testSingleNodeWrite(TransactionalMode.test_unsafe); + withCluster(cluster -> { + testSingleNodeWrite(cluster, TransactionalMode.test_unsafe); + }); } /** * Repair should repair (fully replicate _some_ state) any divergent state between replicas */ @Test - public void fullRepairTest() + public void fullRepairTest() throws Throwable { - testSingleNodeWrite(TransactionalMode.full); + withCluster(cluster -> { + testSingleNodeWrite(cluster, TransactionalMode.full); + }); } @Test - public void onlyAccordTest() + public void onlyAccordTest() throws Throwable + { + withCluster(this::onlyAccordTest); + } + + public void onlyAccordTest(Cluster cluster) { - SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName)); final String keyspace = KEYSPACE; - final String table = accordTableName; + final String table = "accord_table"; + final String qualifiedTable = String.format("%s.%s", keyspace, table); + cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, table)); - executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" + + executeWithRetry(cluster, format("BEGIN TRANSACTION\n" + "INSERT INTO %s (k, v) VALUES (1, 1);\n" + - "COMMIT TRANSACTION", qualifiedAccordTableName)); + "COMMIT TRANSACTION", qualifiedTable)); - SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { + cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table); awaitLocalApplyOnKey(metadata, 1); })); - SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); + cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); - SHARED_CLUSTER.filters().reset(); - awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3)); - nodetool(SHARED_CLUSTER.get(1), "repair", "--accord-only", KEYSPACE); + cluster.filters().reset(); + awaitEndpointUp(cluster.get(1), cluster.get(3)); + nodetool(cluster.get(1), "repair", "--accord-only", KEYSPACE); - SHARED_CLUSTER.get(1).runOnInstance(() -> { + cluster.get(1).runOnInstance(() -> { Assert.assertTrue(barrierRecordingService().executedBarriers); }); } @Test - public void onlyAccordWithForceTest() + public void onlyAccordWithForceTest() throws Throwable + { + withCluster(this::onlyAccordWithForceTest); + } + + public void onlyAccordWithForceTest(Cluster cluster) { - SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName)); final String keyspace = KEYSPACE; - final String table = accordTableName; + final String table = "accord_table"; + final String qualifiedTable = String.format("%s.%s", keyspace, table); + cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, table)); - SHARED_CLUSTER.filters().allVerbs().to(3).drop(); - awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3)); - awaitEndpointDown(SHARED_CLUSTER.get(2), SHARED_CLUSTER.get(3)); + cluster.filters().allVerbs().to(3).drop(); + awaitEndpointDown(cluster.get(1), cluster.get(3)); + awaitEndpointDown(cluster.get(2), cluster.get(3)); - executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" + + executeWithRetry(cluster, format("BEGIN TRANSACTION\n" + "INSERT INTO %s (k, v) VALUES (1, 1);\n" + - "COMMIT TRANSACTION", qualifiedAccordTableName)); + "COMMIT TRANSACTION", qualifiedTable)); - SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { + cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> { TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table); awaitLocalApplyOnKey(metadata, 1); })); - SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); + cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset())); - SHARED_CLUSTER.filters().reset(); - awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3)); - nodetool(SHARED_CLUSTER.get(1), "repair", "--force", "--accord-only", KEYSPACE); + cluster.filters().reset(); + awaitEndpointUp(cluster.get(1), cluster.get(3)); + nodetool(cluster.get(1), "repair", "--force", "--accord-only", KEYSPACE); - SHARED_CLUSTER.get(1).runOnInstance(() -> { + cluster.get(1).runOnInstance(() -> { Assert.assertTrue(barrierRecordingService().executedBarriers); }); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java index 2236f27cd1..6334ff6957 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java @@ -111,6 +111,7 @@ import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.cassandra.db.SystemKeyspace.CONSENSUS_MIGRATION_STATE; import static org.apache.cassandra.db.SystemKeyspace.PAXOS; import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL; +import static org.apache.cassandra.distributed.shared.ClusterUtils.maxEpoch; import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME; import static org.junit.Assert.assertArrayEquals; @@ -163,17 +164,22 @@ public abstract class AccordTestBase extends TestBaseImpl @After public void tearDown() throws Exception { + SHARED_CLUSTER.filters().reset(); for (IInvokableInstance instance : SHARED_CLUSTER) instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start())); truncateSystemTables(); - ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, 1); - SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> ClusterMetadata.current().epoch.getEpoch() == - ((AccordService) AccordService.instance()).configService().currentEpoch() && - AccordService.instance().topology().current().epoch() == - ((AccordService) AccordService.instance()).configService().currentEpoch(), - 60)); + ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, maxEpoch(SHARED_CLUSTER, 1), true); + SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> { + ClusterMetadata metadata = ClusterMetadata.current(); + if (!metadata.schema.hasAccordKeyspaces()) + return true; + + AccordService accord = ((AccordService) AccordService.instance()); + return metadata.epoch.getEpoch() == accord.configService().currentEpoch() && + metadata.epoch.getEpoch() == accord.topology().current().epoch(); + }, 60)); } protected static void assertRowSerial(Cluster cluster, String query, int k, int c, int v, int s) @@ -370,7 +376,7 @@ public abstract class AccordTestBase extends TestBaseImpl instance.runOnInstance(runnable); } - private static Cluster createCluster(int nodes, Function options) throws IOException + public static Cluster createCluster(int nodes, Function options) throws IOException { // need to up the timeout else tests get flaky // disable vnode for now, but should enable before trunk diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java index 4fc4d59534..064d050207 100644 --- a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java +++ b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java @@ -127,7 +127,7 @@ public class JournalGCTest extends FuzzTestBase }, maximumId); int after =-1; - int maxCycles = 3; + int maxCycles = 10; for (int i = 0; i < maxCycles; i++) { cluster.get(1).acceptOnInstance((ks, tbl) -> { diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java index fb9e3f41ae..ada4e51a34 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java @@ -40,6 +40,7 @@ import com.google.common.util.concurrent.FutureCallback; import org.apache.cassandra.auth.PasswordSaltSupplier; import org.apache.cassandra.concurrent.ExecutorFactory; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.ParameterizedClass; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.Constants; @@ -96,6 +97,7 @@ import org.apache.cassandra.simulator.utils.KindOfSequence; import org.apache.cassandra.simulator.utils.LongRange; import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.Closeable; +import org.apache.cassandra.utils.StorageCompatibilityMode; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.Ref; import org.apache.cassandra.utils.memory.BufferPool; @@ -119,6 +121,11 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; @SuppressWarnings("RedundantCast") public class ClusterSimulation implements AutoCloseable { + static + { + CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setEnum(StorageCompatibilityMode.NONE); + } + public static final Class[] SHARE = new Class[] { AsyncFunction.class, diff --git a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java index 5fed781d18..162aa45b2f 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java +++ b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java @@ -40,6 +40,11 @@ import picocli.CommandLine.Command; AccordSimulationRunner.Reconcile.class}) public class AccordSimulationRunner extends SimulationRunner { + static + { + CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setString(StorageCompatibilityMode.NONE.toString()); + } + @BeforeClass public static void beforeAll() { diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java index c139acbe3c..4cb8b2fd0e 100644 --- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java +++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java @@ -18,6 +18,7 @@ package org.apache.cassandra.dht; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Random; @@ -31,6 +32,7 @@ import com.google.common.base.Predicates; import com.google.common.collect.Lists; import com.google.common.collect.Multimap; import org.junit.AfterClass; +import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; @@ -47,7 +49,6 @@ import org.apache.cassandra.gms.IFailureDetector; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.schema.Schema; -import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.StreamOperation; import org.apache.cassandra.tcm.ClusterMetadata; @@ -107,16 +108,16 @@ public class BootStrapperTest @Test public void testSourceTargetComputation() throws UnknownHostException { - final int[] clusterSizes = new int[] { 1, 3, 5, 10, 100 }; + final int[] clusterSizes = new int[]{1, 3, 5, 10, 100}; + List keyspaces = new ArrayList<>(); for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces().names()) { - if (keyspaceName.equals(SchemaConstants.METADATA_KEYSPACE_NAME)) - continue; - int replicationFactor = Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor().allReplicas; - for (int clusterSize : clusterSizes) - if (clusterSize >= replicationFactor) - testSourceTargetComputation(keyspaceName, clusterSize, replicationFactor); + if (keyspaceName.startsWith("BootStrapperTest")) + keyspaces.add(keyspaceName); } + + for (int clusterSize : clusterSizes) + testSourceTargetComputation(keyspaces, clusterSize); } @Test @@ -155,39 +156,47 @@ public class BootStrapperTest } } - private RangeStreamer testSourceTargetComputation(String keyspaceName, int numOldNodes, int replicationFactor) throws UnknownHostException + private void testSourceTargetComputation(List keyspaces, int clusterSize) throws UnknownHostException { ServerTestUtils.resetCMS(); - generateFakeEndpoints(numOldNodes); + generateFakeEndpoints(clusterSize); ClusterMetadata metadata = ClusterMetadata.current(); - assertEquals(numOldNodes, metadata.tokenMap.tokens().size()); + assertEquals(clusterSize, metadata.tokenMap.tokens().size()); + + InetAddressAndPort myEndpoint = InetAddressAndPort.getByName("127.0.0.1"); RangeStreamer s = getRangeStreamer(); + for (String keyspace : keyspaces) + { + assertNotNull(Keyspace.open(keyspace)); + int replicationFactor = Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor().allReplicas; + if (clusterSize < replicationFactor) + continue; - assertNotNull(Keyspace.open(keyspaceName)); - s.addKeyspaceToFetch(keyspaceName); - Multimap toFetch = s.toFetch().get(keyspaceName); - // Pre-TCM this test would always run with RangeStreamer::useStrictSourcesForRanges returning false because - // the RangeStreamer instance was constructed with a null Collection, relying on pending ranges being - // calculated in the test code, and then cloning TokenMetadata with pending tokens added to pass to - // calculateRangesToFetchWithPreferredEndpoints. Post-TCM, we calculate both relaxed and strict movements - // together and TokenMetadata is no more, so the equivalent operation now ends up using strict movements. For - // that reason, when RF includes transient replicas, toFetch will include both a transient and full source, - // hence we dedupe the ranges here. - Set> fetchRanges = toFetch.values() - .stream() - .map(fr -> fr.remote.range()) - .collect(Collectors.toSet()); - // Post CEP-21 wrapping ranges are also unwrapped at this point, so account for that when setting expectation - int expectedRangeCount = replicationFactor + (includesWraparound(fetchRanges) ? 1 : 0); - assertEquals(expectedRangeCount, fetchRanges.size()); + s.addKeyspaceToFetch(keyspace); + Multimap toFetch = s.toFetch().get(keyspace); - // there isn't any point in testing the size of these collections for any specific size. When a random partitioner - // is used, they will vary. - assert toFetch.values().size() > 0; + // Pre-TCM this test would always run with RangeStreamer::useStrictSourcesForRanges returning false because + // the RangeStreamer instance was constructed with a null Collection, relying on pending ranges being + // calculated in the test code, and then cloning TokenMetadata with pending tokens added to pass to + // calculateRangesToFetchWithPreferredEndpoints. Post-TCM, we calculate both relaxed and strict movements + // together and TokenMetadata is no more, so the equivalent operation now ends up using strict movements. For + // that reason, when RF includes transient replicas, toFetch will include both a transient and full source, + // hence we dedupe the ranges here. + Set> fetchRanges = toFetch.values() + .stream() + .map(fr -> fr.remote.range()) + .collect(Collectors.toSet()); + // Post CEP-21 wrapping ranges are also unwrapped at this point, so account for that when setting expectation + int expectedRangeCount = replicationFactor + (includesWraparound(fetchRanges) ? 1 : 0); + assertEquals(expectedRangeCount, fetchRanges.size()); - assert toFetch.keys().stream().noneMatch(InetAddressAndPort.getByName("127.0.0.1")::equals); - return s; + // there isn't any point in testing the size of these collections for any specific size. When a random partitioner + // is used, they will vary. + Assert.assertTrue(toFetch.values().size() > 0); + + Assert.assertTrue(toFetch.keys().stream().noneMatch(myEndpoint::equals)); + } } private RangeStreamer getRangeStreamer() throws UnknownHostException diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java index 77f379571f..7b32878e71 100644 --- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java @@ -29,18 +29,9 @@ import java.util.Set; import java.util.TreeSet; import java.util.function.Supplier; import java.util.stream.Collectors; - import javax.annotation.Nullable; - -import com.google.common.collect.Iterables; -import com.google.common.collect.Sets; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; - import accord.api.Journal; import accord.api.RoutingKey; -import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.CommandStores.RangesForEpoch; import accord.local.DurableBefore; @@ -58,13 +49,13 @@ import accord.primitives.Route; import accord.primitives.SaveStatus; import accord.primitives.Txn; import accord.primitives.TxnId; -import accord.topology.Shard; -import accord.topology.Topology; import accord.utils.Gen; import accord.utils.Gens; import accord.utils.Property.Command; import accord.utils.Property.UnitCommand; import accord.utils.RandomSource; +import com.google.common.collect.Iterables; +import com.google.common.collect.Sets; import org.agrona.collections.Int2ObjectHashMap; import org.agrona.collections.Long2ObjectHashMap; import org.agrona.collections.ObjectHashSet; @@ -83,12 +74,10 @@ import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.dht.Murmur3Partitioner.LongToken; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableId; -import org.apache.cassandra.service.accord.AccordCommandStore; import org.apache.cassandra.service.accord.AccordJournal; import org.apache.cassandra.service.accord.AccordKeyspace; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.AccordTestUtils; -import org.apache.cassandra.service.accord.AccordTopology; import org.apache.cassandra.service.accord.IAccordService; import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo; import org.apache.cassandra.service.accord.TokenRange; @@ -103,12 +92,14 @@ import org.apache.cassandra.utils.RTree; import org.apache.cassandra.utils.RangeTree; import org.apache.cassandra.utils.concurrent.CountDownLatch; import org.assertj.core.api.Assertions; -import org.mockito.Mockito; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + import static accord.local.RedundantStatus.SomeStatus.NONE; import static accord.utils.Property.commands; import static accord.utils.Property.stateful; -import static accord.utils.SortedArrays.SortedArrayList.ofSorted; import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; import static org.apache.cassandra.index.accord.AccordIndexUtil.normalize; import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME; @@ -512,7 +503,6 @@ public class RouteIndexTest extends CQLTester storeRangesForEpochs.put(i, new RangesForEpoch(1, Ranges.of(TokenRange.fullRange(tableId, getPartitioner())))); accordService = startAccord(); - accordService.configService().listener.notifyPostCommit(null, ClusterMetadata.current(), false); accordService.epochReady(ClusterMetadata.current().epoch).awaitUninterruptibly(); minDecidedIdNull = rs.nextFloat(); @@ -525,19 +515,11 @@ public class RouteIndexTest extends CQLTester AccordService startAccord() { - NodeId tcmNodeId = ClusterMetadata.current().myNodeId(); - AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmNodeId)); - Topology topology = new Topology(1, Shard.create(TokenRange.fullRange(tableId, getPartitioner()), ofSorted(new Node.Id(1)), ofSorted(new Node.Id(1)))); - as.unsafeStartupWithOverrides(new Journal.TopologyUpdate(storeRangesForEpochs, topology)); - for (CommandStore commandStore : as.node().commandStores().all()) - ((AccordCommandStore)commandStore).unsafeUpsertRedundantBefore(emptyRedundantBefore); - // the reason for the mocking is to speed up compaction. Collecting the info from the stores has been slow and its always empty in this test... so stub it out to speed up the test - AccordService mock = Mockito.spy(as); - Mockito.doReturn(emptyCompactionInfo(tableId, emptyRedundantBefore, storeRangesForEpochs)).when(mock).getCompactionInfo(); - AccordService.unsafeSetNewAccordService(mock); - - AccordService.replayJournal(as); - return as; + ClusterMetadata metadata = ClusterMetadata.current(); + AccordService.MetadataChangeListener.instance.resetForTesting(metadata); + NodeId tcmNodeId = metadata.myNodeId(); + AccordService.unsafeSetNewAccordService(null); + return AccordService.startup(tcmNodeId); } TxnId nextTxnId(Domain domain) @@ -626,6 +608,7 @@ public class RouteIndexTest extends CQLTester private void restartAccord() { + accordService.journal().closeCurrentSegmentForTestingIfNonEmpty(); accordService.shutdown(); accordService = startAccord(); }