result = apply(view -> updateLiveSet(toRemove, emptySet()).apply(view));
}
}
diff --git a/src/java/org/apache/cassandra/gms/ApplicationState.java b/src/java/org/apache/cassandra/gms/ApplicationState.java
index d31f50cef4..4e20d62048 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
{
// never remove a state here, ordering matters.
@@ -39,6 +47,15 @@ public enum ApplicationState
INTERNAL_ADDRESS_AND_PORT, //Replacement for INTERNAL_IP with up to two ports
NATIVE_ADDRESS_AND_PORT, //Replacement for RPC_ADDRESS
STATUS_WITH_PORT, //Replacement for STATUS
+ /**
+ * 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,
// DO NOT EDIT OR REMOVE PADDING STATES BELOW - only add new states above. See CASSANDRA-16484
X1,
X2,
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
index 1e8604b666..0242d837e1 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
@@ -81,7 +81,7 @@ public class GossipDigestAckVerbHandler extends GossipVerbHandler deltaEpStateMap = new HashMap();
+ Map deltaEpStateMap = new HashMap<>();
for (GossipDigest gDigest : gDigestList)
{
InetAddressAndPort addr = gDigest.getEndpoint();
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index 7720379b8d..a092c77007 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -105,9 +105,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
SILENT_SHUTDOWN_STATES.add(VersionedValue.STATUS_BOOTSTRAPPING);
SILENT_SHUTDOWN_STATES.add(VersionedValue.STATUS_BOOTSTRAPPING_REPLACE);
}
- private static List ADMINISTRATIVELY_INACTIVE_STATES = Arrays.asList(VersionedValue.HIBERNATE,
- VersionedValue.REMOVED_TOKEN,
- VersionedValue.STATUS_LEFT);
+ private static final List ADMINISTRATIVELY_INACTIVE_STATES = Arrays.asList(VersionedValue.HIBERNATE,
+ VersionedValue.REMOVED_TOKEN,
+ VersionedValue.STATUS_LEFT);
private volatile ScheduledFuture scheduledGossipTask;
private static final ReentrantLock taskLock = new ReentrantLock();
public final static int intervalInMillis = 1000;
@@ -124,11 +124,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
// Maximimum difference between generation value and local time we are willing to accept about a peer
static final int MAX_GENERATION_DIFFERENCE = 86400 * 365;
- private long fatClientTimeout;
+ private final long fatClientTimeout;
private final Random random = new Random();
/* subscribers for interest in EndpointState change */
- private final List subscribers = new CopyOnWriteArrayList();
+ private final List subscribers = new CopyOnWriteArrayList<>();
/* live member set */
@VisibleForTesting
@@ -258,7 +258,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat();
if (logger.isTraceEnabled())
logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion());
- final List gDigests = new ArrayList();
+ final List gDigests = new ArrayList<>();
Gossiper.instance.makeRandomGossipDigest(gDigests);
if (gDigests.size() > 0)
@@ -662,7 +662,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
for (GossipDigest gDigest : gDigests)
{
sb.append(gDigest);
- sb.append(" ");
+ sb.append(' ');
}
logger.trace("Gossip Digests are : {}", sb);
}
@@ -741,7 +741,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
InetAddressAndPort endpoint = InetAddressAndPort.getByName(address);
runInGossipStageBlocking(() -> {
EndpointState epState = endpointStateMap.get(endpoint);
- Collection tokens = null;
+ Collection tokens;
logger.warn("Assassinating {} via gossip", endpoint);
if (epState == null)
@@ -1076,6 +1076,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(InetAddressAndPort 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(InetAddressAndPort forEndpoint, int version)
{
EndpointState epState = endpointStateMap.get(forEndpoint);
@@ -1653,7 +1671,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public void start(int generationNumber)
{
- start(generationNumber, new EnumMap(ApplicationState.class));
+ start(generationNumber, new EnumMap<>(ApplicationState.class));
}
/**
@@ -1717,7 +1735,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
seedsInShadowRound.clear();
endpointShadowStateMap.clear();
// send a completely empty syn
- List gDigests = new ArrayList();
+ List gDigests = new ArrayList<>();
GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),
DatabaseDescriptor.getPartitionerName(),
gDigests);
@@ -1835,7 +1853,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public List getSeeds()
{
- List seedList = new ArrayList();
+ List seedList = new ArrayList<>();
for (InetAddressAndPort seed : seeds)
{
seedList.add(seed.toString());
@@ -2074,7 +2092,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public Map> getReleaseVersionsWithPort()
{
- Map> results = new HashMap>();
+ Map> results = new HashMap<>();
Iterable allHosts = Iterables.concat(Gossiper.instance.getLiveMembers(), Gossiper.instance.getUnreachableMembers());
for (InetAddressAndPort host : allHosts)
diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java
index 7c545598bf..880cb98e06 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;
@@ -31,6 +33,7 @@ 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.VersionAndType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
@@ -109,7 +112,7 @@ public class VersionedValue implements Comparable
@Override
public String toString()
{
- return "Value(" + value + "," + version + ")";
+ return "Value(" + value + ',' + version + ')';
}
public byte[] toBytes()
@@ -298,6 +301,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 dc674513ad..b781ebf50c 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -80,6 +80,12 @@ public class Descriptor
this(formatType.info.getLatestVersion(), directory, ksname, cfname, generation, formatType);
}
+ @VisibleForTesting
+ public Descriptor(String version, File directory, String ksname, String cfname, int generation, SSTableFormat.Type formatType)
+ {
+ this(formatType.info.getVersion(version), directory, ksname, cfname, generation, formatType);
+ }
+
public Descriptor(Version version, File directory, String ksname, String cfname, int generation, SSTableFormat.Type formatType)
{
assert version != null && directory != null && ksname != null && cfname != null && formatType.info.getLatestVersion().getClass().equals(version.getClass());
@@ -354,6 +360,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/Version.java b/src/java/org/apache/cassandra/io/sstable/format/Version.java
index 0e9e303faf..501ae85149 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/Version.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/Version.java
@@ -97,7 +97,6 @@ public abstract class Version
return version;
}
-
@Override
public boolean equals(Object o)
{
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/schema/Schema.java b/src/java/org/apache/cassandra/schema/Schema.java
index c04c6311e5..c5c1f36d5d 100644
--- a/src/java/org/apache/cassandra/schema/Schema.java
+++ b/src/java/org/apache/cassandra/schema/Schema.java
@@ -17,6 +17,7 @@
*/
package org.apache.cassandra.schema;
+import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
@@ -578,7 +579,7 @@ public final class Schema implements SchemaProvider
updateVersionAndAnnounce();
}
- public synchronized TransformationResult transform(SchemaTransformation transformation, boolean locally, long now)
+ public synchronized TransformationResult transform(SchemaTransformation transformation, boolean locally, long now) throws UnknownHostException
{
KeyspacesDiff diff;
try
diff --git a/src/java/org/apache/cassandra/schema/SchemaTransformation.java b/src/java/org/apache/cassandra/schema/SchemaTransformation.java
index c19ac7c7c7..e2290a339b 100644
--- a/src/java/org/apache/cassandra/schema/SchemaTransformation.java
+++ b/src/java/org/apache/cassandra/schema/SchemaTransformation.java
@@ -17,6 +17,8 @@
*/
package org.apache.cassandra.schema;
+import java.net.UnknownHostException;
+
public interface SchemaTransformation
{
/**
@@ -27,5 +29,5 @@ public interface SchemaTransformation
* @param schema Keyspaces to base the transformation on
* @return Keyspaces transformed by the statement
*/
- Keyspaces apply(Keyspaces schema);
+ Keyspaces apply(Keyspaces schema) throws UnknownHostException;
}
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/StartupChecks.java b/src/java/org/apache/cassandra/service/StartupChecks.java
index 85b5836baf..dadb0c5b8d 100644
--- a/src/java/org/apache/cassandra/service/StartupChecks.java
+++ b/src/java/org/apache/cassandra/service/StartupChecks.java
@@ -36,10 +36,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import net.jpountz.lz4.LZ4Factory;
-import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
-import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -55,8 +53,10 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.NativeLibrary;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JavaUtils;
+import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.SigarLibrary;
+import static java.lang.String.format;
import static org.apache.cassandra.config.CassandraRelevantProperties.COM_SUN_MANAGEMENT_JMXREMOTE_PORT;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VERSION;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VM_NAME;
@@ -83,7 +83,6 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_VM_NA
public class StartupChecks
{
private static final Logger logger = LoggerFactory.getLogger(StartupChecks.class);
-
// List of checks to run before starting up. If any test reports failure, startup will be halted.
private final List preFlightChecks = new ArrayList<>();
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 8c5437c2ee..e0ac1b6b36 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -83,6 +83,8 @@ import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.io.sstable.SSTableLoader;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+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;
@@ -281,6 +283,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
private final StreamStateStore streamStateStore = new StreamStateStore();
+ public final SSTablesGlobalTracker sstablesTracker;
+
public boolean isSurveyMode()
{
return isSurveyMode;
@@ -323,6 +327,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
jmxObjectName = "org.apache.cassandra.db:type=StorageService";
MBeanWrapper.instance.registerMBean(this, jmxObjectName);
MBeanWrapper.instance.registerMBean(StreamManager.instance, StreamManager.OBJECT_NAME);
+
+ sstablesTracker = new SSTablesGlobalTracker(SSTableFormat.Type.current());
}
public void registerDaemon(CassandraDaemon daemon)
@@ -946,11 +952,24 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
appStates.put(ApplicationState.NATIVE_ADDRESS_AND_PORT, valueFactory.nativeaddressAndPort(FBUtilities.getBroadcastNativeAddressAndPort()));
appStates.put(ApplicationState.RPC_ADDRESS, valueFactory.rpcaddress(FBUtilities.getJustBroadcastNativeAddress()));
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.
gossipActive = true;
+
+ 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/CompactStorage3to4UpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage3to4UpgradeTest.java
index e94c2c4917..8d051d384a 100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage3to4UpgradeTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage3to4UpgradeTest.java
@@ -22,6 +22,9 @@ import org.junit.Test;
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.apache.cassandra.distributed.shared.AssertUtils.assertRows;
public class CompactStorage3to4UpgradeTest extends UpgradeTestBase
@@ -33,6 +36,7 @@ public class CompactStorage3to4UpgradeTest extends UpgradeTestBase
{
new TestCase().nodes(1)
.upgrade(Versions.Major.v30, Versions.Major.v4)
+ .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL))
.setup(cluster -> {
String create = "CREATE TABLE %s.%s(k int, c1 int, c2 int, v int, PRIMARY KEY (k, c1, c2)) " +
"WITH compaction = { 'class':'LeveledCompactionStrategy', 'enabled':'false'} AND COMPACT STORAGE";
diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/CompactTableTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/CompactTableTest.java
index 46a7b1d57a..bac94a1cbf 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/CompactTableTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/CompactTableTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.cql3.validation.operations;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
index 910445f5b6..4390b20781 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
@@ -89,14 +89,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());
@@ -150,12 +150,17 @@ public class TrackerTest
{
ColumnFamilyStore cfs = MockSchema.newCFS(metadata -> metadata.caching(CachingParams.CACHE_KEYS));
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.isKeyCacheEnabled());
@@ -184,6 +189,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);
@@ -239,15 +245,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();
}
@@ -344,7 +351,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);
@@ -369,7 +376,7 @@ public class TrackerTest
MockListener failListener = new MockListener(true);
tracker.subscribe(failListener);
tracker.subscribe(listener);
- Assert.assertNotNull(tracker.notifyAdded(singleton(r1), null, null));
+ Assert.assertNotNull(tracker.notifyAdded(singleton(r1), false, null, null));
Assert.assertEquals(singleton(r1), ((SSTableAddedNotification) listener.received.get(0)).added);
Assert.assertFalse(((SSTableAddedNotification) listener.received.get(0)).memtable().isPresent());
listener.received.clear();
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..e4a5947b4d
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/SSTablesGlobalTrackerTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.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()
+ {
+ return Generate.enumValues(SSTableFormat.Type.class);
+ }
+
+ 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(SSTableFormat.Type.current().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