repairStatusesChanged)
diff --git a/src/java/org/apache/cassandra/gms/ApplicationState.java b/src/java/org/apache/cassandra/gms/ApplicationState.java
index ade9208e05..70e6d9ce8f 100644
--- a/src/java/org/apache/cassandra/gms/ApplicationState.java
+++ b/src/java/org/apache/cassandra/gms/ApplicationState.java
@@ -17,6 +17,14 @@
*/
package org.apache.cassandra.gms;
+/**
+ * The various "states" exchanged through Gossip.
+ *
+ * Important Note: Gossip uses the ordinal of this enum in the messages it exchanges, so values in that enum
+ * should not be re-ordered or removed. The end of this enum should also always include some "padding" so that
+ * if newer versions add new states, old nodes that don't know about those new states don't "break" deserializing those
+ * states.
+ */
public enum ApplicationState
{
STATUS,
@@ -34,11 +42,22 @@ public enum ApplicationState
HOST_ID,
TOKENS,
RPC_READY,
- // pad to allow adding new states to existing cluster
+ // We added SSTABLE_VERSIONS in CASSANDRA-15897 in 3.0, and at the time, 3 more ApplicationState had been added
+ // to newer versions, so we skipped the first 3 of our original padding to ensure SSTABLE_VERSIONS can preserve
+ // its ordinal accross versions.
X1,
X2,
X3,
- X4,
+ /**
+ * The set of sstable versions on this node. This will usually be only the "current" sstable format (the one with
+ * which new sstables are written), but may contain more on newly upgraded nodes before `upgradesstable` has been
+ * run.
+ *
+ *
The value (a set of sstable {@link org.apache.cassandra.io.sstable.format.VersionAndType}) is serialized as
+ * a comma-separated list.
+ **/
+ SSTABLE_VERSIONS,
+ // pad to allow adding new states to existing cluster
X5,
X6,
X7,
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index b09d9e196a..0f37dc96f8 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -935,6 +935,24 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return UUID.fromString(epStates.get(endpoint).getApplicationState(ApplicationState.HOST_ID).value);
}
+ /**
+ * The value for the provided application state for the provided endpoint as currently known by this Gossip instance.
+ *
+ * @param endpoint the endpoint from which to get the endpoint state.
+ * @param state the endpoint state to get.
+ * @return the value of the application state {@code state} for {@code endpoint}, or {@code null} if either
+ * {@code endpoint} is not known by Gossip or has no value for {@code state}.
+ */
+ public String getApplicationState(InetAddress endpoint, ApplicationState state)
+ {
+ EndpointState epState = endpointStateMap.get(endpoint);
+ if (epState == null)
+ return null;
+
+ VersionedValue value = epState.getApplicationState(state);
+ return value == null ? null : value.value;
+ }
+
EndpointState getStateForVersionBiggerThan(InetAddress forEndpoint, int version)
{
EndpointState epState = endpointStateMap.get(forEndpoint);
diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java
index d9c8d0bc50..2d345b9cfa 100644
--- a/src/java/org/apache/cassandra/gms/VersionedValue.java
+++ b/src/java/org/apache/cassandra/gms/VersionedValue.java
@@ -20,7 +20,9 @@ package org.apache.cassandra.gms;
import java.io.*;
import java.net.InetAddress;
import java.util.Collection;
+import java.util.Set;
import java.util.UUID;
+import java.util.stream.Collectors;
import static java.nio.charset.StandardCharsets.ISO_8859_1;
@@ -30,6 +32,8 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.sstable.format.Version;
+import org.apache.cassandra.io.sstable.format.VersionAndType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
@@ -267,6 +271,13 @@ public class VersionedValue implements Comparable
{
return new VersionedValue(String.valueOf(value));
}
+
+ public VersionedValue sstableVersions(Set versions)
+ {
+ return new VersionedValue(versions.stream()
+ .map(VersionAndType::toString)
+ .collect(Collectors.joining(",")));
+ }
}
private static class VersionedValueSerializer implements IVersionedSerializer
diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
index 811e1a1f16..3828a15aca 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -359,6 +359,7 @@ public class Descriptor
&& that.generation == this.generation
&& that.ksname.equals(this.ksname)
&& that.cfname.equals(this.cfname)
+ && that.version.equals(this.version)
&& that.formatType == this.formatType;
}
diff --git a/src/java/org/apache/cassandra/io/sstable/format/VersionAndType.java b/src/java/org/apache/cassandra/io/sstable/format/VersionAndType.java
new file mode 100644
index 0000000000..7d698a9980
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/sstable/format/VersionAndType.java
@@ -0,0 +1,93 @@
+/*
+ * 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.io.sstable.format;
+
+import java.util.List;
+import java.util.Objects;
+
+import com.google.common.base.Splitter;
+
+import org.apache.cassandra.io.sstable.Descriptor;
+
+/**
+ * Groups a sstable {@link Version} with a {@link SSTableFormat.Type}.
+ *
+ * Note that both information are currently necessary to identify the exact "format" of an sstable (without having
+ * its {@link Descriptor}). In particular, while {@link Version} contains its {{@link SSTableFormat}}, you cannot get
+ * the {{@link SSTableFormat.Type}} from that.
+ */
+public final class VersionAndType
+{
+ private static final Splitter splitOnDash = Splitter.on('-').omitEmptyStrings().trimResults();
+
+ private final Version version;
+ private final SSTableFormat.Type formatType;
+
+ public VersionAndType(Version version, SSTableFormat.Type formatType)
+ {
+ this.version = version;
+ this.formatType = formatType;
+ }
+
+ public Version version()
+ {
+ return version;
+ }
+
+ public SSTableFormat.Type formatType()
+ {
+ return formatType;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ VersionAndType that = (VersionAndType) o;
+ return Objects.equals(version, that.version) &&
+ formatType == that.formatType;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(version, formatType);
+ }
+
+ public static VersionAndType fromString(String versionAndType)
+ {
+ List components = splitOnDash.splitToList(versionAndType);
+ if (components.size() != 2)
+ throw new IllegalArgumentException("Invalid VersionAndType string: " + versionAndType + " (should be of the form 'big-bc')");
+
+ SSTableFormat.Type formatType = SSTableFormat.Type.validate(components.get(0));
+ Version version = formatType.info.getVersion(components.get(1));
+ return new VersionAndType(version, formatType);
+ }
+
+ @Override
+ public String toString()
+ {
+ return formatType.name + '-' + version;
+ }
+}
diff --git a/src/java/org/apache/cassandra/notifications/InitialSSTableAddedNotification.java b/src/java/org/apache/cassandra/notifications/InitialSSTableAddedNotification.java
new file mode 100644
index 0000000000..db66496be1
--- /dev/null
+++ b/src/java/org/apache/cassandra/notifications/InitialSSTableAddedNotification.java
@@ -0,0 +1,36 @@
+/*
+ * 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.notifications;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+
+/**
+ * Fired when we load SSTables during the Cassandra initialization phase.
+ */
+public class InitialSSTableAddedNotification implements INotification
+{
+ /** {@code true} if the addition corresponds to the {@link ColumnFamilyStore} initialization, then the sstables
+ * are those loaded at startup. */
+ public final Iterable added;
+
+ public InitialSSTableAddedNotification(Iterable added)
+ {
+ this.added = added;
+ }
+}
diff --git a/src/java/org/apache/cassandra/notifications/SSTableAddedNotification.java b/src/java/org/apache/cassandra/notifications/SSTableAddedNotification.java
index 56d61303f5..81e8cc3200 100644
--- a/src/java/org/apache/cassandra/notifications/SSTableAddedNotification.java
+++ b/src/java/org/apache/cassandra/notifications/SSTableAddedNotification.java
@@ -22,6 +22,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
public class SSTableAddedNotification implements INotification
{
public final Iterable added;
+
public SSTableAddedNotification(Iterable added)
{
this.added = added;
diff --git a/src/java/org/apache/cassandra/service/SSTablesGlobalTracker.java b/src/java/org/apache/cassandra/service/SSTablesGlobalTracker.java
new file mode 100644
index 0000000000..de78892c9e
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/SSTablesGlobalTracker.java
@@ -0,0 +1,284 @@
+/*
+ * 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.service;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.lifecycle.Tracker;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.io.sstable.format.VersionAndType;
+import org.apache.cassandra.notifications.INotification;
+import org.apache.cassandra.notifications.INotificationConsumer;
+import org.apache.cassandra.notifications.InitialSSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableDeletingNotification;
+import org.apache.cassandra.notifications.SSTableListChangedNotification;
+import org.apache.cassandra.utils.NoSpamLogger;
+
+/**
+ * Tracks all sstables in use on the local node.
+ *
+ * Each table tracks its own SSTables in {@link ColumnFamilyStore} (through {@link Tracker}) for most purposes, but
+ * this class groups information we need on all the sstables the node has.
+ */
+public class SSTablesGlobalTracker implements INotificationConsumer
+{
+ private static final Logger logger = LoggerFactory.getLogger(SSTablesGlobalTracker.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 5L, TimeUnit.MINUTES);
+
+ /*
+ * As of CASSANDRA-15897, the only thing we track here is the set of sstable versions in use.
+ *
+ * That set is maintained in `versionsInUse`, an immutable set replaced when changed (so that it can be read safely
+ * and cheaply). To track when that set changes and needs to be re-computed, we essentially maintain a map of
+ * sstables versions in use to the number of sstables for that version. But as we know that sstables will
+ * overwhelmingly be on the "current" version, we special case said current version (makes things cheaper without
+ * too much complexity here). Those are `sstablesForCurrentVersion` and `sstablesForOtherVersions`.
+ *
+ * This would be sufficient if we could guarantee that for any sstable, we only ever have 1 addition notification
+ * and 1 corresponding remove notification. But sstables can have complex lifecycles and relying on this property
+ * could be fragile. As a matter of fact, at the time of this writing, the removal notification is sometimes fired
+ * twice for the same sstable. To keep this component more resilient, we also maintain the set of all sstables for
+ * which we've received an addition, which allows us to ignore removals for sstables we don't know.
+ *
+ * Concurrency handling: the 'allSSTables' set handles concurrency directly as it is updated in all cases. The rest
+ * of the data structures of this class are only updated together within a synchronized block when handling new
+ * sstables additions/removals.
+ */
+
+ private final Set allSSTables = ConcurrentHashMap.newKeySet();
+
+ private final VersionAndType currentVersion;
+ private int sstablesForCurrentVersion;
+ private final Map sstablesForOtherVersions = new HashMap<>();
+
+ private volatile ImmutableSet versionsInUse = ImmutableSet.of();
+
+ private final Set subscribers = new CopyOnWriteArraySet<>();
+
+ public SSTablesGlobalTracker(SSTableFormat.Type currentSSTableFormat)
+ {
+ this.currentVersion = new VersionAndType(currentSSTableFormat.info.getLatestVersion(), currentSSTableFormat);
+ }
+
+ /**
+ * The set of all sstable versions currently in use on this node.
+ */
+ public Set versionsInUse()
+ {
+ return versionsInUse;
+ }
+
+ /**
+ * Register a new subscriber to this tracker.
+ *
+ * Registered subscribers are currently notified when the set of sstable versions in use changes, using a
+ * {@link SSTablesVersionsInUseChangeNotification}.
+ *
+ * @param subscriber the new subscriber to register. If this subscriber is already registered, this method does
+ * nothing (meaning that even if a subscriber is registered multiple times, it will only be notified once on every
+ * change).
+ * @return whether the subscriber was register (so whether it was not already registered).
+ */
+ public boolean register(INotificationConsumer subscriber)
+ {
+ return subscribers.add(subscriber);
+ }
+
+ /**
+ * Unregister a subscriber from this tracker.
+ *
+ * @param subscriber the subscriber to unregister. If this subscriber is not registered, this method does nothing.
+ * @return whether the subscriber was unregistered (so whether it was registered subscriber of this tracker).
+ */
+ public boolean unregister(INotificationConsumer subscriber)
+ {
+ return subscribers.remove(subscriber);
+ }
+
+ @Override
+ public void handleNotification(INotification notification, Object sender)
+ {
+ Iterable removed = removedSSTables(notification);
+ Iterable added = addedSSTables(notification);
+ if (Iterables.isEmpty(removed) && Iterables.isEmpty(added))
+ return;
+
+ boolean triggerUpdate = handleSSTablesChange(removed, added);
+ if (triggerUpdate)
+ {
+ SSTablesVersionsInUseChangeNotification changeNotification = new SSTablesVersionsInUseChangeNotification(versionsInUse);
+ subscribers.forEach(s -> s.handleNotification(changeNotification, this));
+ }
+ }
+
+ @VisibleForTesting
+ boolean handleSSTablesChange(Iterable removed, Iterable added)
+ {
+ /*
+ We collect changes to 'sstablesForCurrentVersion' and 'sstablesForOtherVersions' as delta first, and then
+ apply those delta within a synchronized block below. The goal being to reduce the work done in that
+ synchronized block.
+ */
+ int currentDelta = 0;
+ Map othersDelta = null;
+ /*
+ Note: we deal with removes first as if a notification both removes and adds, it's a compaction and while
+ it should never remove and add the same descriptor in practice, doing the remove first is more logical.
+ */
+ for (Descriptor desc : removed)
+ {
+ if (!allSSTables.remove(desc))
+ continue;
+
+ VersionAndType version = version(desc);
+ if (currentVersion.equals(version))
+ --currentDelta;
+ else
+ othersDelta = update(othersDelta, version, -1);
+ }
+ for (Descriptor desc : added)
+ {
+ if (!allSSTables.add(desc))
+ continue;
+
+ VersionAndType version = version(desc);
+ if (currentVersion.equals(version))
+ ++currentDelta;
+ else
+ othersDelta = update(othersDelta, version, +1);
+ }
+
+ if (currentDelta == 0 && (othersDelta == null))
+ return false;
+
+ /*
+ Set to true if the set of versions in use is changed by this update. That is, if a version having no
+ version prior now has some, or if the count for some version reaches 0.
+ */
+ boolean triggerUpdate;
+ synchronized (this)
+ {
+ triggerUpdate = (currentDelta > 0 && sstablesForCurrentVersion == 0)
+ || (currentDelta < 0 && sstablesForCurrentVersion <= -currentDelta);
+ sstablesForCurrentVersion += currentDelta;
+ sstablesForCurrentVersion = sanitizeSSTablesCount(sstablesForCurrentVersion, currentVersion);
+
+ if (othersDelta != null)
+ {
+ for (Map.Entry entry : othersDelta.entrySet())
+ {
+ VersionAndType version = entry.getKey();
+ int delta = entry.getValue();
+ /*
+ Updates the count, removing the version if it reaches 0 (note: we could use Map#compute for this,
+ but we wouldn't be able to modify `triggerUpdate` without making it an Object, so we don't bother).
+ */
+ Integer oldValue = sstablesForOtherVersions.get(version);
+ int newValue = oldValue == null ? delta : oldValue + delta;
+ newValue = sanitizeSSTablesCount(newValue, version);
+ triggerUpdate |= oldValue == null || newValue == 0;
+ if (newValue == 0)
+ sstablesForOtherVersions.remove(version);
+ else
+ sstablesForOtherVersions.put(version, newValue);
+ }
+ }
+
+ if (triggerUpdate)
+ versionsInUse = computeVersionsInUse(sstablesForCurrentVersion, currentVersion, sstablesForOtherVersions);
+ }
+ return triggerUpdate;
+ }
+
+ private static ImmutableSet computeVersionsInUse(int sstablesForCurrentVersion, VersionAndType currentVersion, Map sstablesForOtherVersions)
+ {
+ ImmutableSet.Builder builder = ImmutableSet.builder();
+ if (sstablesForCurrentVersion > 0)
+ builder.add(currentVersion);
+ builder.addAll(sstablesForOtherVersions.keySet());
+ return builder.build();
+ }
+
+ private static int sanitizeSSTablesCount(int sstableCount, VersionAndType version)
+ {
+ if (sstableCount >= 0)
+ return sstableCount;
+
+ /*
+ This shouldn't happen and indicate a bug either in the tracking of this class, or on the passed notification.
+ That said, it's not worth bringing the node down, so we log the problem but otherwise "correct" it.
+ */
+ noSpamLogger.error("Invalid state while handling sstables change notification: the number of sstables for " +
+ "version {} was computed to {}. This indicate a bug and please report it, but it should " +
+ "not have adverse consequences.", version, sstableCount, new RuntimeException());
+ return 0;
+ }
+
+ private static Iterable addedSSTables(INotification notification)
+ {
+ if (notification instanceof SSTableAddedNotification)
+ return Iterables.transform(((SSTableAddedNotification)notification).added, s -> s.descriptor);
+ if (notification instanceof SSTableListChangedNotification)
+ return Iterables.transform(((SSTableListChangedNotification)notification).added, s -> s.descriptor);
+ if (notification instanceof InitialSSTableAddedNotification)
+ return Iterables.transform(((InitialSSTableAddedNotification)notification).added, s -> s.descriptor);
+ else
+ return Collections.emptyList();
+ }
+
+ private static Iterable removedSSTables(INotification notification)
+ {
+ if (notification instanceof SSTableDeletingNotification)
+ return Collections.singletonList(((SSTableDeletingNotification)notification).deleting.descriptor);
+ if (notification instanceof SSTableListChangedNotification)
+ return Iterables.transform(((SSTableListChangedNotification)notification).removed, s -> s.descriptor);
+ else
+ return Collections.emptyList();
+ }
+
+ private static Map update(Map counts,
+ VersionAndType toUpdate,
+ int delta)
+ {
+ Map m = counts == null ? new HashMap<>() : counts;
+ m.merge(toUpdate, delta, (a, b) -> (a + b == 0) ? null : (a + b));
+ return m;
+ }
+
+ @VisibleForTesting
+ static VersionAndType version(Descriptor sstable)
+ {
+ return new VersionAndType(sstable.version, sstable.formatType);
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/SSTablesVersionsInUseChangeNotification.java b/src/java/org/apache/cassandra/service/SSTablesVersionsInUseChangeNotification.java
new file mode 100644
index 0000000000..352bec7ba8
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/SSTablesVersionsInUseChangeNotification.java
@@ -0,0 +1,50 @@
+/*
+ * 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.service;
+
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.cassandra.io.sstable.format.VersionAndType;
+import org.apache.cassandra.notifications.INotification;
+
+/**
+ * Notification triggered by a {@link SSTablesGlobalTracker} when the set of sstables versions in use on this node
+ * changes.
+ *
+ * The notification includes the set of sstable versions in use when the notification is triggered (so the result
+ * of the change triggering that notification).
+ */
+public class SSTablesVersionsInUseChangeNotification implements INotification
+{
+ /**
+ * The set of all sstable versions in use on this node at the time of this notification.
+ */
+ public final ImmutableSet versionsInUse;
+
+ SSTablesVersionsInUseChangeNotification(ImmutableSet versionsInUse)
+ {
+ this.versionsInUse = versionsInUse;
+ }
+
+ @Override
+ public String toString()
+ {
+ return String.format("SSTablesInUseChangeNotification(%s)", versionsInUse);
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 31db74f06d..906854c163 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -73,6 +73,7 @@ import org.apache.cassandra.gms.*;
import org.apache.cassandra.hints.HintVerbHandler;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.io.sstable.SSTableLoader;
+import org.apache.cassandra.io.sstable.format.VersionAndType;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.*;
import org.apache.cassandra.metrics.StorageMetrics;
@@ -244,6 +245,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
private final StreamStateStore streamStateStore = new StreamStateStore();
+ public final SSTablesGlobalTracker sstablesTracker;
+
public boolean isSurveyMode()
{
return isSurveyMode;
@@ -319,6 +322,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.BATCH_STORE, new BatchStoreVerbHandler());
MessagingService.instance().registerVerbHandlers(MessagingService.Verb.BATCH_REMOVE, new BatchRemoveVerbHandler());
+
+ sstablesTracker = new SSTablesGlobalTracker(DatabaseDescriptor.getSSTableFormat());
}
public void registerDaemon(CassandraDaemon daemon)
@@ -842,9 +847,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
appStates.put(ApplicationState.HOST_ID, valueFactory.hostId(localHostId));
appStates.put(ApplicationState.RPC_ADDRESS, valueFactory.rpcaddress(FBUtilities.getBroadcastRpcAddress()));
appStates.put(ApplicationState.RELEASE_VERSION, valueFactory.releaseVersion());
+ appStates.put(ApplicationState.SSTABLE_VERSIONS, valueFactory.sstableVersions(sstablesTracker.versionsInUse()));
+
logger.info("Starting up server gossip");
Gossiper.instance.register(this);
Gossiper.instance.start(SystemKeyspace.incrementAndGetGeneration(), appStates); // needed for node-ring gathering.
+
+ sstablesTracker.register((notification, o) -> {
+ if (!(notification instanceof SSTablesVersionsInUseChangeNotification))
+ return;
+
+ Set versions = ((SSTablesVersionsInUseChangeNotification)notification).versionsInUse;
+ logger.debug("Updating local sstables version in Gossip to {}", versions);
+
+ Gossiper.instance.addLocalApplicationState(ApplicationState.SSTABLE_VERSIONS,
+ valueFactory.sstableVersions(versions));
+ });
+
// gossip snitch infos (local DC and rack)
gossipSnitchInfo();
// gossip Schema.emptyVersion forcing immediate check for schema updates (see MigrationManager#maybeScheduleSchemaPull)
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java
index 4f5d1bcdb1..032befadbd 100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage2to3UpgradeTest.java
@@ -33,6 +33,10 @@ import org.apache.cassandra.distributed.shared.Versions;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
import static org.junit.Assert.assertEquals;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+
public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
{
@@ -54,7 +58,7 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
for (int i = 0; i < cluster.size(); i++)
{
int nodeNum = i + 1;
- System.out.println(String.format("****** node %s: %s", nodeNum, cluster.get(nodeNum).config()));
+ System.out.printf("****** node %s: %s%n", nodeNum, cluster.get(nodeNum).config());
}
})
.runAfterNodeUpgrade(((cluster, node) -> {
@@ -88,7 +92,7 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
for (int i = 0; i < cluster.size(); i++)
{
int nodeNum = i + 1;
- System.out.println(String.format("****** node %s: %s", nodeNum, cluster.get(nodeNum).config()));
+ System.out.printf("****** node %s: %s%n", nodeNum, cluster.get(nodeNum).config());
}
})
.runAfterNodeUpgrade(((cluster, node) -> {
@@ -116,6 +120,7 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
new TestCase()
.nodes(2)
.upgrade(Versions.Major.v22, Versions.Major.v30)
+ .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL))
.setup(cluster -> {
cluster.schemaChange(String.format(
"CREATE TABLE %s.%s (key int, c1 int, c2 int, c3 int, PRIMARY KEY (key, c1, c2)) WITH COMPACT STORAGE",
@@ -159,13 +164,15 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
KEYSPACE, table, partitions - 8, rowsPerPartition - 3),
});
+
}).runAfterClusterUpgrade(cluster ->
{
for (int i = 1; i <= cluster.size(); i++)
{
- NodeToolResult result = cluster.get(1).nodetoolResult("upgradesstables");
+ NodeToolResult result = cluster.get(i).nodetoolResult("upgradesstables");
assertEquals("upgrade sstables failed for node " + i, 0, result.getRc());
}
+ Thread.sleep(600);
// make sure the results are the same after upgrade and upgrade sstables but before dropping compact storage
recorder.validateResults(cluster, 1);
@@ -200,6 +207,7 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
new TestCase()
.nodes(2)
.upgrade(Versions.Major.v22, Versions.Major.v30)
+ .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL))
.setup(cluster -> {
cluster.schemaChange(String.format(
"CREATE TABLE %s.%s (key int, c1 int, c2 int, c3 int, PRIMARY KEY (key, c1, c2)) WITH COMPACT STORAGE",
@@ -218,15 +226,10 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
KEYSPACE, table, i, j), ConsistencyLevel.ALL);
}
}
-
})
.runAfterClusterUpgrade(cluster -> {
- for (int i = 1; i <= cluster.size(); i++)
- {
- NodeToolResult result = cluster.get(1).nodetoolResult("upgradesstables");
- assertEquals("upgrade sstables failed for node " + i, 0, result.getRc());
- }
-
+ cluster.forEach(n -> n.nodetoolResult("upgradesstables", KEYSPACE).asserts().success());
+ Thread.sleep(1000);
// drop compact storage on only one node before performing writes
IMessageFilters.Filter filter = cluster.verbs().allVerbs().to(2).drop();
cluster.schemaChange(String.format("ALTER TABLE %s.%s DROP COMPACT STORAGE", KEYSPACE, table), 1);
@@ -262,7 +265,7 @@ public class CompactStorage2to3UpgradeTest extends UpgradeTestBase
KEYSPACE, table, 8, 1, 3), ConsistencyLevel.ALL);
coordinator.execute(String.format("DELETE FROM %s.%s WHERE key = %d and c1 = %d and c2 > 1",
- KEYSPACE, table, 6, 2, 4), ConsistencyLevel.ALL);
+ KEYSPACE, table, 6, 2), ConsistencyLevel.ALL);
ResultsRecorder recorder = new ResultsRecorder();
runQueries(coordinator, recorder, new String[] {
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/DropCompactStorageTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/DropCompactStorageTest.java
new file mode 100644
index 0000000000..8207448579
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/DropCompactStorageTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.upgrade;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.shared.Versions;
+
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+public class DropCompactStorageTest extends UpgradeTestBase
+{
+ @Test
+ public void dropCompactStorageBeforeUpgradesstablesTo30() throws Throwable
+ {
+ dropCompactStorageBeforeUpgradeSstables(Versions.Major.v30);
+ }
+
+ /**
+ * Upgrades a node from 2.2 to 3.x and DROP COMPACT just after the upgrade but _before_ upgrading the underlying
+ * sstables.
+ *
+ * This test reproduces the issue from CASSANDRA-15897.
+ */
+ public void dropCompactStorageBeforeUpgradeSstables(Versions.Major upgradeTo) throws Throwable
+ {
+ new TestCase()
+ .nodes(1)
+ .upgrade(Versions.Major.v22, upgradeTo)
+ .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL))
+ .setup((cluster) -> {
+ cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (id int, ck int, v int, PRIMARY KEY (id, ck)) WITH COMPACT STORAGE");
+ for (int i = 0; i < 5; i++)
+ cluster.coordinator(1).execute("INSERT INTO "+KEYSPACE+".tbl (id, ck, v) values (1, ?, ?)", ConsistencyLevel.ALL, i, i);
+ cluster.get(1).flush(KEYSPACE);
+ })
+ .runAfterNodeUpgrade((cluster, node) -> {
+ Throwable thrown = catchThrowable(() -> cluster.schemaChange("ALTER TABLE "+KEYSPACE+".tbl DROP COMPACT STORAGE"));
+ assertThat(thrown).hasMessageContainingAll("Cannot DROP COMPACT STORAGE as some nodes in the cluster",
+ "has some non-upgraded 2.x sstables");
+
+ assertThat(cluster.get(1).nodetool("upgradesstables")).isEqualTo(0);
+ cluster.schemaChange("ALTER TABLE "+KEYSPACE+".tbl DROP COMPACT STORAGE");
+ cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.ALL);
+ })
+ .run();
+ }
+}
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
index de1e640148..ddae1fcef1 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
@@ -86,14 +86,14 @@ public class TrackerTest
List readers = ImmutableList.of(MockSchema.sstable(0, true, cfs), MockSchema.sstable(1, cfs), MockSchema.sstable(2, cfs));
tracker.addInitialSSTables(copyOf(readers));
Assert.assertNull(tracker.tryModify(ImmutableList.of(MockSchema.sstable(0, cfs)), OperationType.COMPACTION));
- try (LifecycleTransaction txn = tracker.tryModify(readers.get(0), OperationType.COMPACTION);)
+ try (LifecycleTransaction txn = tracker.tryModify(readers.get(0), OperationType.COMPACTION))
{
Assert.assertNotNull(txn);
Assert.assertNull(tracker.tryModify(readers.get(0), OperationType.COMPACTION));
Assert.assertEquals(1, txn.originals().size());
Assert.assertTrue(txn.originals().contains(readers.get(0)));
}
- try (LifecycleTransaction txn = tracker.tryModify(Collections.emptyList(), OperationType.COMPACTION);)
+ try (LifecycleTransaction txn = tracker.tryModify(Collections.emptyList(), OperationType.COMPACTION))
{
Assert.assertNotNull(txn);
Assert.assertEquals(0, txn.originals().size());
@@ -147,12 +147,17 @@ public class TrackerTest
{
ColumnFamilyStore cfs = MockSchema.newCFS();
Tracker tracker = cfs.getTracker();
+ MockListener listener = new MockListener(false);
+ tracker.subscribe(listener);
List readers = ImmutableList.of(MockSchema.sstable(0, 17, cfs),
MockSchema.sstable(1, 121, cfs),
MockSchema.sstable(2, 9, cfs));
tracker.addInitialSSTables(copyOf(readers));
Assert.assertEquals(3, tracker.view.get().sstables.size());
+ Assert.assertEquals(1, listener.senders.size());
+ Assert.assertEquals(1, listener.received.size());
+ Assert.assertTrue(listener.received.get(0) instanceof InitialSSTableAddedNotification);
for (SSTableReader reader : readers)
Assert.assertTrue(reader.isKeyCacheSetup());
@@ -181,6 +186,7 @@ public class TrackerTest
Assert.assertEquals(17 + 121 + 9, cfs.metric.liveDiskSpaceUsed.getCount());
Assert.assertEquals(1, listener.senders.size());
+ Assert.assertEquals(1, listener.received.size());
Assert.assertEquals(tracker, listener.senders.get(0));
Assert.assertTrue(listener.received.get(0) instanceof SSTableAddedNotification);
DatabaseDescriptor.setIncrementalBackupsEnabled(backups);
@@ -236,15 +242,16 @@ public class TrackerTest
Assert.assertNull(tracker.dropSSTables(reader -> reader != readers.get(0), OperationType.UNKNOWN, null));
Assert.assertEquals(1, tracker.getView().sstables.size());
- Assert.assertEquals(3, listener.received.size());
+ Assert.assertEquals(4, listener.received.size());
Assert.assertEquals(tracker, listener.senders.get(0));
- Assert.assertTrue(listener.received.get(0) instanceof SSTableDeletingNotification);
- Assert.assertTrue(listener.received.get(1) instanceof SSTableDeletingNotification);
- Assert.assertTrue(listener.received.get(2) instanceof SSTableListChangedNotification);
- Assert.assertEquals(readers.get(1), ((SSTableDeletingNotification) listener.received.get(0)).deleting);
- Assert.assertEquals(readers.get(2), ((SSTableDeletingNotification)listener.received.get(1)).deleting);
- Assert.assertEquals(2, ((SSTableListChangedNotification) listener.received.get(2)).removed.size());
- Assert.assertEquals(0, ((SSTableListChangedNotification) listener.received.get(2)).added.size());
+ Assert.assertTrue(listener.received.get(0) instanceof InitialSSTableAddedNotification);
+ Assert.assertTrue(listener.received.get(1) instanceof SSTableDeletingNotification);
+ Assert.assertTrue(listener.received.get(2) instanceof SSTableDeletingNotification);
+ Assert.assertTrue(listener.received.get(3) instanceof SSTableListChangedNotification);
+ Assert.assertEquals(readers.get(1), ((SSTableDeletingNotification) listener.received.get(1)).deleting);
+ Assert.assertEquals(readers.get(2), ((SSTableDeletingNotification)listener.received.get(2)).deleting);
+ Assert.assertEquals(2, ((SSTableListChangedNotification) listener.received.get(3)).removed.size());
+ Assert.assertEquals(0, ((SSTableListChangedNotification) listener.received.get(3)).added.size());
Assert.assertEquals(9, cfs.metric.liveDiskSpaceUsed.getCount());
readers.get(0).selfRef().release();
}
@@ -335,7 +342,7 @@ public class TrackerTest
Tracker tracker = new Tracker(null, false);
MockListener listener = new MockListener(false);
tracker.subscribe(listener);
- tracker.notifyAdded(singleton(r1));
+ tracker.notifyAdded(singleton(r1), false);
Assert.assertEquals(singleton(r1), ((SSTableAddedNotification) listener.received.get(0)).added);
listener.received.clear();
tracker.notifyDeleting(r1);
@@ -356,7 +363,7 @@ public class TrackerTest
MockListener failListener = new MockListener(true);
tracker.subscribe(failListener);
tracker.subscribe(listener);
- Assert.assertNotNull(tracker.notifyAdded(singleton(r1), null));
+ Assert.assertNotNull(tracker.notifyAdded(singleton(r1), false, null));
Assert.assertEquals(singleton(r1), ((SSTableAddedNotification) listener.received.get(0)).added);
listener.received.clear();
Assert.assertNotNull(tracker.notifySSTablesChanged(singleton(r1), singleton(r2), OperationType.COMPACTION, null));
diff --git a/test/unit/org/apache/cassandra/io/sstable/format/VersionAndTypeTest.java b/test/unit/org/apache/cassandra/io/sstable/format/VersionAndTypeTest.java
new file mode 100644
index 0000000000..633993f0f9
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/format/VersionAndTypeTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.io.sstable.format;
+
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static junit.framework.Assert.assertEquals;
+
+public class VersionAndTypeTest
+{
+ @Test
+ public void testValidInput()
+ {
+ assertEquals(VersionAndType.fromString("big-bc").toString(), "big-bc");
+ }
+
+ @Test
+ public void testInvalidInputs()
+ {
+ assertThatThrownBy(() -> VersionAndType.fromString(" ")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be of the form 'big-bc'");
+ assertThatThrownBy(() -> VersionAndType.fromString("mcc-")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be of the form 'big-bc'");
+ assertThatThrownBy(() -> VersionAndType.fromString("mcc-d")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("No Type constant mcc");
+ assertThatThrownBy(() -> VersionAndType.fromString("mcc-d-")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("No Type constant mcc");
+ assertThatThrownBy(() -> VersionAndType.fromString("mcc")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be of the form 'big-bc'");
+ assertThatThrownBy(() -> VersionAndType.fromString("-")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be of the form 'big-bc'");
+ assertThatThrownBy(() -> VersionAndType.fromString("--")).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be of the form 'big-bc'");
+ }
+}
diff --git a/test/unit/org/apache/cassandra/service/SSTablesGlobalTrackerTest.java b/test/unit/org/apache/cassandra/service/SSTablesGlobalTrackerTest.java
new file mode 100644
index 0000000000..cde7242562
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/SSTablesGlobalTrackerTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.service;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.io.sstable.format.VersionAndType;
+import org.assertj.core.util.Files;
+import org.quicktheories.core.Gen;
+import org.quicktheories.generators.Generate;
+
+import static org.junit.Assert.assertEquals;
+import static org.quicktheories.QuickTheory.qt;
+import static org.quicktheories.generators.SourceDSL.integers;
+import static org.quicktheories.generators.SourceDSL.lists;
+import static org.quicktheories.generators.SourceDSL.strings;
+
+public class SSTablesGlobalTrackerTest
+{
+ private static final int MAX_VERSION_LIST_SIZE = 10;
+ private static final int MAX_UPDATES_PER_GEN = 100;
+
+ /**
+ * Ensures that the tracker properly maintains the set of versions in use.
+ *
+ * Using 'Quick Theories', we generate a number of random sstables notification and validate that after each
+ * update the tracker computes the proper set of versions in use (using a simplisitic model that keeps the set
+ * of all sstables in use and maps it to the set of its versions).
+ */
+ @Test
+ public void testUpdates()
+ {
+ qt().forAll(lists().of(updates()).ofSizeBetween(0, MAX_UPDATES_PER_GEN),
+ sstableFormatTypes())
+ .checkAssert((updates, formatType) -> {
+ SSTablesGlobalTracker tracker = new SSTablesGlobalTracker(formatType);
+ Set all = new HashSet<>();
+ Set previous = Collections.emptySet();
+ for (Update update : updates)
+ {
+ update.applyTo(all);
+ boolean triggerUpdate = tracker.handleSSTablesChange(update.removed, update.added);
+ Set expectedInUse = versionAndTypes(all);
+ assertEquals(expectedInUse, tracker.versionsInUse());
+ assertEquals(!expectedInUse.equals(previous), triggerUpdate);
+ previous = expectedInUse;
+ }
+ });
+ }
+
+ private Set versionAndTypes(Set descriptors)
+ {
+ return descriptors.stream().map(SSTablesGlobalTracker::version).collect(Collectors.toSet());
+ }
+
+ private Gen keyspaces()
+ {
+ return Generate.pick(Arrays.asList("k1", "k2"));
+ }
+
+ private Gen tables()
+ {
+ return Generate.pick(Arrays.asList("t1", "t2", "t3"));
+ }
+
+ private Gen generations()
+ {
+ return integers().between(1, 20);
+ }
+
+ private Gen descriptors()
+ {
+ return sstableFormatTypes().zip(keyspaces(),
+ tables(),
+ generations(),
+ sstableVersionString(),
+ (f, k, t, g, v) -> new Descriptor(v, Files.currentFolder(), k, t, g, f));
+ }
+
+ private Gen> descriptorLists(int minSize)
+ {
+ return lists().of(descriptors()).ofSizeBetween(minSize, MAX_VERSION_LIST_SIZE);
+ }
+
+ private Gen sstableFormatTypes()
+ {
+ // We remap LEGACY format to BIG as is not used in real life and will not work properly because the
+ // SSTableFormat.Type.validate() method never returns it.
+ return Generate.enumValues(SSTableFormat.Type.class)
+ .map(t -> t == SSTableFormat.Type.LEGACY ? SSTableFormat.Type.BIG : t);
+ }
+
+ private Gen sstableVersionString()
+ {
+ // We want to somewhat favor the current version, as that is technically more realistic so we generate it 50%
+ // of the time, and generate something random 50% of the time.
+ return Generate.constant(DatabaseDescriptor.getSSTableFormat().info.getLatestVersion().getVersion())
+ .mix(strings().betweenCodePoints('a', 'z').ofLength(2));
+ }
+
+ private Gen updates()
+ {
+ // We want to avoid update that remove and add nothing, as those appear to be generated quite a bit are not
+ // too useful. Yet, having one of removed/added be empty is actually something we want.
+ Gen maybeEmptyRemoved = descriptorLists(0).zip(descriptorLists(1), Update::new);
+ Gen maybeEmptyAdded = descriptorLists(1).zip(descriptorLists(0), Update::new);
+ return maybeEmptyRemoved.mix(maybeEmptyAdded);
+ }
+
+ private static class Update
+ {
+ final List removed;
+ final List added;
+
+ Update(List removed, List added)
+ {
+ this.removed = removed;
+ this.added = added;
+ }
+
+ void applyTo(Set allSSTables)
+ {
+ allSSTables.removeAll(removed);
+ allSSTables.addAll(added);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Update{" +
+ "removed=" + removed +
+ ", added=" + added +
+ '}';
+ }
+ }
+}
\ No newline at end of file