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

Merge branch 'cassandra-3.11' into trunk

This commit is contained in:
Ekaterina Dimitrova 2021-03-10 20:54:07 -05:00
parent 0e990d7231 b654000936
commit c7432e98a3
25 changed files with 919 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -37,6 +37,7 @@ Merged from 3.11:
* Reduce amount of allocations during batch statement execution (CASSANDRA-16201)
* Update jflex-1.6.0.jar to match upstream (CASSANDRA-16393)
Merged from 3.0:
* Refuse DROP COMPACT STORAGE if some 2.x sstables are in use (CASSANDRA-15897)
* Fix ColumnFilter::toString not returning a valid CQL fragment (CASSANDRA-16483)
* Fix ColumnFilter behaviour to prevent digest mitmatches during upgrades (CASSANDRA-16415)
* Update debian packaging for python3 (CASSANDRA-16396)

2
build.xml
View File

@ -594,6 +594,7 @@
<dependency groupId="com.google.code.java-allocation-instrumenter" artifactId="java-allocation-instrumenter" version="${allocation-instrumenter.version}" />
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.7" />
<dependency groupId="org.reflections" artifactId="reflections" version="0.9.12" />
<dependency groupId="org.quicktheories" artifactId="quicktheories" version="0.25" />
<dependency groupId="org.apache.rat" artifactId="apache-rat" version="0.10">
<exclusion groupId="commons-lang" artifactId="commons-lang"/>
</dependency>
@ -682,7 +683,6 @@
<!-- when updating assertj, make sure to also update the corresponding junit-bom dependency -->
<dependency groupId="org.assertj" artifactId="assertj-core" version="3.15.0"/>
<dependency groupId="org.awaitility" artifactId="awaitility" version="4.0.3" />
</dependencyManagement>
<developer id="adelapena" name="Andres de la Peña"/>
<developer id="alakshman" name="Avinash Lakshman"/>

93
src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.cql3.statements.schema;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@ -24,9 +25,14 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogContext;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.Permission;
@ -38,6 +44,11 @@ import org.apache.cassandra.cql3.QualifiedName;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.KeyspaceMetadata;
@ -48,11 +59,14 @@ import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.schema.ViewMetadata;
import org.apache.cassandra.schema.Views;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.transport.Event.SchemaChange;
import org.apache.cassandra.transport.Event.SchemaChange.Change;
import org.apache.cassandra.transport.Event.SchemaChange.Target;
import org.apache.cassandra.utils.NoSpamLogger;
import static java.lang.String.format;
import static java.lang.String.join;
import static com.google.common.collect.Iterables.isEmpty;
@ -68,7 +82,7 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
this.tableName = tableName;
}
public Keyspaces apply(Keyspaces schema)
public Keyspaces apply(Keyspaces schema) throws UnknownHostException
{
KeyspaceMetadata keyspace = schema.getNullable(keyspaceName);
@ -103,10 +117,10 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
public String toString()
{
return String.format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, tableName);
return format("%s (%s, %s)", getClass().getSimpleName(), keyspaceName, tableName);
}
abstract KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table);
abstract KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table) throws UnknownHostException;
/**
* ALTER TABLE <table> ALTER <column> TYPE <newtype>;
@ -404,6 +418,8 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
*/
private static class DropCompactStorage extends AlterTableStatement
{
private static final Logger logger = LoggerFactory.getLogger(AlterTableStatement.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 5L, TimeUnit.MINUTES);
private DropCompactStorage(String keyspaceName, String tableName)
{
super(keyspaceName, tableName);
@ -414,8 +430,79 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
if (!table.isCompactTable())
throw AlterTableStatement.ire("Cannot DROP COMPACT STORAGE on table without COMPACT STORAGE");
validateCanDropCompactStorage();
return keyspace.withSwapped(keyspace.tables.withSwapped(table.withSwapped(ImmutableSet.of(TableMetadata.Flag.COMPOUND))));
}
/**
* Throws if DROP COMPACT STORAGE cannot be used (yet) because the cluster is not sufficiently upgraded. To be able
* to use DROP COMPACT STORAGE, we need to ensure that no pre-3.0 sstables exists in the cluster, as we won't be
* able to read them anymore once COMPACT STORAGE is dropped (see CASSANDRA-15897). In practice, this method checks
* 3 things:
* 1) that all nodes are on 3.0+. We need this because 2.x nodes don't advertise their sstable versions.
* 2) for 3.0+, we use the new (CASSANDRA-15897) sstables versions set gossiped by all nodes to ensure all
* sstables have been upgraded cluster-wise.
* 3) if the cluster still has some 3.0 nodes that predate CASSANDRA-15897, we will not have the sstable versions
* for them. In that case, we also refuse DROP COMPACT (even though it may well be safe at this point) and ask
* the user to upgrade all nodes.
*/
private void validateCanDropCompactStorage()
{
Set<InetAddressAndPort> before4 = new HashSet<>();
Set<InetAddressAndPort> preC15897nodes = new HashSet<>();
Set<InetAddressAndPort> with2xSStables = new HashSet<>();
Splitter onComma = Splitter.on(',').omitEmptyStrings().trimResults();
for (InetAddressAndPort node : StorageService.instance.getTokenMetadata().getAllEndpoints())
{
if (MessagingService.instance().versions.knows(node) &&
MessagingService.instance().versions.getRaw(node) < MessagingService.VERSION_40)
{
before4.add(node);
continue;
}
String sstableVersionsString = Gossiper.instance.getApplicationState(node, ApplicationState.SSTABLE_VERSIONS);
if (sstableVersionsString == null)
{
preC15897nodes.add(node);
continue;
}
try
{
boolean has2xSStables = onComma.splitToList(sstableVersionsString)
.stream()
.anyMatch(v -> v.compareTo("big-ma")<=0);
if (has2xSStables)
with2xSStables.add(node);
}
catch (IllegalArgumentException e)
{
// Means VersionType::fromString didn't parse a version correctly. Which shouldn't happen, we shouldn't
// have garbage in Gossip. But crashing the request is not ideal, so we log the error but ignore the
// node otherwise.
noSpamLogger.error("Unexpected error parsing sstable versions from gossip for {} (gossiped value " +
"is '{}'). This is a bug and should be reported. Cannot ensure that {} has no " +
"non-upgraded 2.x sstables anymore. If after this DROP COMPACT STORAGE some old " +
"sstables cannot be read anymore, please use `upgradesstables` with the " +
"`--force-compact-storage-on` option.", node, sstableVersionsString, node);
}
}
if (!before4.isEmpty())
throw new InvalidRequestException(format("Cannot DROP COMPACT STORAGE as some nodes in the cluster (%s) " +
"are not on 4.0+ yet. Please upgrade those nodes and run " +
"`upgradesstables` before retrying.", before4));
if (!preC15897nodes.isEmpty())
throw new InvalidRequestException(format("Cannot guarantee that DROP COMPACT STORAGE is safe as some nodes " +
"in the cluster (%s) do not have https://issues.apache.org/jira/browse/CASSANDRA-15897. " +
"Please upgrade those nodes and retry.", preC15897nodes));
if (!with2xSStables.isEmpty())
throw new InvalidRequestException(format("Cannot DROP COMPACT STORAGE as some nodes in the cluster (%s) " +
"has some non-upgraded 2.x sstables. Please run `upgradesstables` " +
"on those nodes before retrying", with2xSStables));
}
}
public static final class Raw extends CQLStatement.Raw

4
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -390,6 +390,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
initialMemtable = new Memtable(new AtomicReference<>(CommitLog.instance.getCurrentPosition()), this);
data = new Tracker(initialMemtable, loadSSTables);
// Note that this needs to happen before we load the first sstables, or the global sstable tracker will not
// be notified on the initial loading.
data.subscribe(StorageService.instance.sstablesTracker);
Collection<SSTableReader> sstables = null;
// scan for sstables corresponding to this cf and load them
if (data.loadsstables)

5
src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
View File

@ -125,7 +125,7 @@ public class CompactionStrategyManager implements INotificationConsumer
private volatile CompactionParams params;
private DiskBoundaries currentBoundaries;
private volatile boolean enabled = true;
private volatile boolean enabled;
private volatile boolean isActive = true;
/*
@ -747,7 +747,8 @@ public class CompactionStrategyManager implements INotificationConsumer
if (notification instanceof SSTableAddedNotification)
{
handleFlushNotification(((SSTableAddedNotification) notification).added);
SSTableAddedNotification flushedNotification = (SSTableAddedNotification) notification;
handleFlushNotification(flushedNotification.added);
}
else if (notification instanceof SSTableListChangedNotification)
{

61
src/java/org/apache/cassandra/db/lifecycle/Tracker.java
View File

@ -107,7 +107,7 @@ public class Tracker
Pair<View, View> apply(Function<View, View> function)
{
return apply(Predicates.<View>alwaysTrue(), function);
return apply(Predicates.alwaysTrue(), function);
}
Throwable apply(Function<View, View> function, Throwable accumulate)
@ -186,17 +186,12 @@ public class Tracker
public void addInitialSSTables(Iterable<SSTableReader> sstables)
{
addInitialSSTablesWithoutUpdatingSize(sstables);
maybeFail(updateSizeTracking(emptySet(), sstables, null));
// no notifications or backup necessary
addSSTablesInternal(sstables, true, false, true);
}
public void addInitialSSTablesWithoutUpdatingSize(Iterable<SSTableReader> sstables)
{
if (!isDummy())
setupOnline(sstables);
apply(updateLiveSet(emptySet(), sstables));
// no notifications or backup necessary
addSSTablesInternal(sstables, true, false, false);
}
public void updateInitialSSTableSize(Iterable<SSTableReader> sstables)
@ -206,9 +201,22 @@ public class Tracker
public void addSSTables(Iterable<SSTableReader> sstables)
{
addInitialSSTables(sstables);
maybeIncrementallyBackup(sstables);
notifyAdded(sstables);
addSSTablesInternal(sstables, false, true, true);
}
private void addSSTablesInternal(Iterable<SSTableReader> sstables,
boolean isInitialSSTables,
boolean maybeIncrementallyBackup,
boolean updateSize)
{
if (!isDummy())
setupOnline(sstables);
apply(updateLiveSet(emptySet(), sstables));
if(updateSize)
maybeFail(updateSizeTracking(emptySet(), sstables, null));
if (maybeIncrementallyBackup)
maybeIncrementallyBackup(sstables);
notifyAdded(sstables, isInitialSSTables);
}
/** (Re)initializes the tracker, purging all references. */
@ -236,7 +244,7 @@ public class Tracker
public Throwable dropSSTables(Throwable accumulate)
{
return dropSSTables(Predicates.<SSTableReader>alwaysTrue(), OperationType.UNKNOWN, accumulate);
return dropSSTables(Predicates.alwaysTrue(), OperationType.UNKNOWN, accumulate);
}
/**
@ -267,7 +275,7 @@ public class Tracker
accumulate = updateSizeTracking(removed, emptySet(), accumulate);
accumulate = release(selfRefs(removed), accumulate);
// notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
accumulate = notifySSTablesChanged(removed, Collections.<SSTableReader>emptySet(), txnLogs.type(), accumulate);
accumulate = notifySSTablesChanged(removed, Collections.emptySet(), txnLogs.type(), accumulate);
}
}
catch (Throwable t)
@ -291,13 +299,7 @@ public class Tracker
*/
public void removeUnreadableSSTables(final File directory)
{
maybeFail(dropSSTables(new Predicate<SSTableReader>()
{
public boolean apply(SSTableReader reader)
{
return reader.descriptor.directory.equals(directory);
}
}, OperationType.UNKNOWN, null));
maybeFail(dropSSTables(reader -> reader.descriptor.directory.equals(directory), OperationType.UNKNOWN, null));
}
@ -371,7 +373,7 @@ public class Tracker
notifyDiscarded(memtable);
// TODO: if we're invalidated, should we notifyadded AND removed, or just skip both?
fail = notifyAdded(sstables, memtable, fail);
fail = notifyAdded(sstables, false, memtable, fail);
if (!isDummy() && !cfstore.isValid())
dropSSTables();
@ -429,9 +431,14 @@ public class Tracker
return accumulate;
}
Throwable notifyAdded(Iterable<SSTableReader> added, Memtable memtable, Throwable accumulate)
Throwable notifyAdded(Iterable<SSTableReader> added, boolean isInitialSSTables, Memtable memtable, Throwable accumulate)
{
INotification notification = new SSTableAddedNotification(added, memtable);
INotification notification;
if (!isInitialSSTables)
notification = new SSTableAddedNotification(added, memtable);
else
notification = new InitialSSTableAddedNotification(added);
for (INotificationConsumer subscriber : subscribers)
{
try
@ -446,9 +453,9 @@ public class Tracker
return accumulate;
}
public void notifyAdded(Iterable<SSTableReader> added)
void notifyAdded(Iterable<SSTableReader> added, boolean isInitialSSTables)
{
maybeFail(notifyAdded(added, null, null));
maybeFail(notifyAdded(added, isInitialSSTables, null, null));
}
public void notifySSTableRepairedStatusChanged(Collection<SSTableReader> repairStatusesChanged)
@ -529,8 +536,6 @@ public class Tracker
@VisibleForTesting
public void removeUnsafe(Set<SSTableReader> toRemove)
{
Pair<View, View> result = apply(view -> {
return updateLiveSet(toRemove, emptySet()).apply(view);
});
Pair<View, View> result = apply(view -> updateLiveSet(toRemove, emptySet()).apply(view));
}
}

17
src/java/org/apache/cassandra/gms/ApplicationState.java
View File

@ -17,6 +17,14 @@
*/
package org.apache.cassandra.gms;
/**
* The various "states" exchanged through Gossip.
*
* <p><b>Important Note:</b> Gossip uses the ordinal of this enum in the messages it exchanges, so values in that enum
* should <i>not</i> 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.
*
* <p>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,

2
src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
View File

@ -81,7 +81,7 @@ public class GossipDigestAckVerbHandler extends GossipVerbHandler<GossipDigestAc
}
/* Get the state required to send to this gossipee - construct GossipDigestAck2Message */
Map<InetAddressAndPort, EndpointState> deltaEpStateMap = new HashMap<InetAddressAndPort, EndpointState>();
Map<InetAddressAndPort, EndpointState> deltaEpStateMap = new HashMap<>();
for (GossipDigest gDigest : gDigestList)
{
InetAddressAndPort addr = gDigest.getEndpoint();

42
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -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<String> ADMINISTRATIVELY_INACTIVE_STATES = Arrays.asList(VersionedValue.HIBERNATE,
VersionedValue.REMOVED_TOKEN,
VersionedValue.STATUS_LEFT);
private static final List<String> 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<IEndpointStateChangeSubscriber> subscribers = new CopyOnWriteArrayList<IEndpointStateChangeSubscriber>();
private final List<IEndpointStateChangeSubscriber> 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<GossipDigest> gDigests = new ArrayList<GossipDigest>();
final List<GossipDigest> 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<Token> tokens = null;
Collection<Token> 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, VersionedValue>(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<GossipDigest> gDigests = new ArrayList<GossipDigest>();
List<GossipDigest> gDigests = new ArrayList<>();
GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),
DatabaseDescriptor.getPartitionerName(),
gDigests);
@ -1835,7 +1853,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public List<String> getSeeds()
{
List<String> seedList = new ArrayList<String>();
List<String> seedList = new ArrayList<>();
for (InetAddressAndPort seed : seeds)
{
seedList.add(seed.toString());
@ -2074,7 +2092,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public Map<String, List<String>> getReleaseVersionsWithPort()
{
Map<String, List<String>> results = new HashMap<String, List<String>>();
Map<String, List<String>> results = new HashMap<>();
Iterable<InetAddressAndPort> allHosts = Iterables.concat(Gossiper.instance.getLiveMembers(), Gossiper.instance.getUnreachableMembers());
for (InetAddressAndPort host : allHosts)

12
src/java/org/apache/cassandra/gms/VersionedValue.java
View File

@ -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<VersionedValue>
@Override
public String toString()
{
return "Value(" + value + "," + version + ")";
return "Value(" + value + ',' + version + ')';
}
public byte[] toBytes()
@ -298,6 +301,13 @@ public class VersionedValue implements Comparable<VersionedValue>
{
return new VersionedValue(String.valueOf(value));
}
public VersionedValue sstableVersions(Set<VersionAndType> versions)
{
return new VersionedValue(versions.stream()
.map(VersionAndType::toString)
.collect(Collectors.joining(",")));
}
}
private static class VersionedValueSerializer implements IVersionedSerializer<VersionedValue>

7
src/java/org/apache/cassandra/io/sstable/Descriptor.java
View File

@ -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;
}

1
src/java/org/apache/cassandra/io/sstable/format/Version.java
View File

@ -97,7 +97,6 @@ public abstract class Version
return version;
}
@Override
public boolean equals(Object o)
{

93
src/java/org/apache/cassandra/io/sstable/format/VersionAndType.java Normal file
View File

@ -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}.
*
* <p>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<String> 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;
}
}

36
src/java/org/apache/cassandra/notifications/InitialSSTableAddedNotification.java Normal file
View File

@ -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<SSTableReader> added;
public InitialSSTableAddedNotification(Iterable<SSTableReader> added)
{
this.added = added;
}
}

3
src/java/org/apache/cassandra/schema/Schema.java
View File

@ -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

4
src/java/org/apache/cassandra/schema/SchemaTransformation.java
View File

@ -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;
}

284
src/java/org/apache/cassandra/service/SSTablesGlobalTracker.java Normal file
View File

@ -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.
*
* <p>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<Descriptor> allSSTables = ConcurrentHashMap.newKeySet();
private final VersionAndType currentVersion;
private int sstablesForCurrentVersion;
private final Map<VersionAndType, Integer> sstablesForOtherVersions = new HashMap<>();
private volatile ImmutableSet<VersionAndType> versionsInUse = ImmutableSet.of();
private final Set<INotificationConsumer> 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<VersionAndType> 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<Descriptor> removed = removedSSTables(notification);
Iterable<Descriptor> 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<Descriptor> removed, Iterable<Descriptor> 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<VersionAndType, Integer> 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<VersionAndType, Integer> 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<VersionAndType> computeVersionsInUse(int sstablesForCurrentVersion, VersionAndType currentVersion, Map<VersionAndType, Integer> sstablesForOtherVersions)
{
ImmutableSet.Builder<VersionAndType> 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<Descriptor> 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<Descriptor> 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<VersionAndType, Integer> update(Map<VersionAndType, Integer> counts,
VersionAndType toUpdate,
int delta)
{
Map<VersionAndType, Integer> 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);
}
}

50
src/java/org/apache/cassandra/service/SSTablesVersionsInUseChangeNotification.java Normal file
View File

@ -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.
*
* <p>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<VersionAndType> versionsInUse;
SSTablesVersionsInUseChangeNotification(ImmutableSet<VersionAndType> versionsInUse)
{
this.versionsInUse = versionsInUse;
}
@Override
public String toString()
{
return String.format("SSTablesInUseChangeNotification(%s)", versionsInUse);
}
}

5
src/java/org/apache/cassandra/service/StartupChecks.java
View File

@ -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<StartupCheck> preFlightChecks = new ArrayList<>();

19
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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<VersionAndType> 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)

4
test/distributed/org/apache/cassandra/distributed/upgrade/CompactStorage3to4UpgradeTest.java
View File

@ -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";

1
test/unit/org/apache/cassandra/cql3/validation/operations/CompactTableTest.java
View File

@ -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;

31
test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
View File

@ -89,14 +89,14 @@ public class TrackerTest
List<SSTableReader> 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.<SSTableReader>emptyList(), OperationType.COMPACTION);)
try (LifecycleTransaction txn = tracker.tryModify(Collections.<SSTableReader>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<SSTableReader> 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();

51
test/unit/org/apache/cassandra/io/sstable/format/VersionAndTypeTest.java Normal file
View File

@ -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'");
}
}

158
test/unit/org/apache/cassandra/service/SSTablesGlobalTrackerTest.java Normal file
View File

@ -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.
*
* <p>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<Descriptor> all = new HashSet<>();
Set<VersionAndType> previous = Collections.emptySet();
for (Update update : updates)
{
update.applyTo(all);
boolean triggerUpdate = tracker.handleSSTablesChange(update.removed, update.added);
Set<VersionAndType> expectedInUse = versionAndTypes(all);
assertEquals(expectedInUse, tracker.versionsInUse());
assertEquals(!expectedInUse.equals(previous), triggerUpdate);
previous = expectedInUse;
}
});
}
private Set<VersionAndType> versionAndTypes(Set<Descriptor> descriptors)
{
return descriptors.stream().map(SSTablesGlobalTracker::version).collect(Collectors.toSet());
}
private Gen<String> keyspaces()
{
return Generate.pick(Arrays.asList("k1", "k2"));
}
private Gen<String> tables()
{
return Generate.pick(Arrays.asList("t1", "t2", "t3"));
}
private Gen<Integer> generations()
{
return integers().between(1, 20);
}
private Gen<Descriptor> descriptors()
{
return sstableFormatTypes().zip(keyspaces(),
tables(),
generations(),
sstableVersionString(),
(f, k, t, g, v) -> new Descriptor(v, Files.currentFolder(), k, t, g, f));
}
private Gen<List<Descriptor>> descriptorLists(int minSize)
{
return lists().of(descriptors()).ofSizeBetween(minSize, MAX_VERSION_LIST_SIZE);
}
private Gen<SSTableFormat.Type> sstableFormatTypes()
{
return Generate.enumValues(SSTableFormat.Type.class);
}
private Gen<String> 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<Update> 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<Update> maybeEmptyRemoved = descriptorLists(0).zip(descriptorLists(1), Update::new);
Gen<Update> maybeEmptyAdded = descriptorLists(1).zip(descriptorLists(0), Update::new);
return maybeEmptyRemoved.mix(maybeEmptyAdded);
}
private static class Update
{
final List<Descriptor> removed;
final List<Descriptor> added;
Update(List<Descriptor> removed, List<Descriptor> added)
{
this.removed = removed;
this.added = added;
}
void applyTo(Set<Descriptor> allSSTables)
{
allSSTables.removeAll(removed);
allSSTables.addAll(added);
}
@Override
public String toString()
{
return "Update{" +
"removed=" + removed +
", added=" + added +
'}';
}
}
}