You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jl...@apache.org on 2023/03/31 15:41:38 UTC
[cassandra] branch cassandra-4.1 updated (780f8b94ff -> 14936d0bd3)
This is an automated email from the ASF dual-hosted git repository.
jlewandowski pushed a change to branch cassandra-4.1
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from 780f8b94ff Merge branch 'cassandra-4.0' into cassandra-4.1
add f96659c530 Save host id to system.local and flush immediately after startup
add ee586b53c2 Merge branch 'cassandra-3.0' into cassandra-3.11
add 4b9c18235a Merge branch 'cassandra-3.11' into cassandra-4.0
new 14936d0bd3 Merge branch 'cassandra-4.0' into cassandra-4.1
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 5 +-
.../org/apache/cassandra/db/SystemKeyspace.java | 23 ++-
.../apache/cassandra/db/commitlog/CommitLog.java | 47 ++++--
.../cassandra/db/compaction/CompactionManager.java | 76 +++++++--
.../apache/cassandra/service/StorageService.java | 173 ++++++++++++++++-----
.../cassandra/tools/SSTableMetadataViewer.java | 31 ++--
.../cassandra/distributed/impl/Instance.java | 12 +-
.../cassandra/distributed/impl/InstanceConfig.java | 4 +-
.../distributed/test/IPMembershipTest.java | 4 +
.../distributed/test/SSTableIdGenerationTest.java | 24 +--
.../unit/org/apache/cassandra/db/KeyCacheTest.java | 2 -
.../cassandra/db/compaction/NeverPurgeTest.java | 8 +-
12 files changed, 302 insertions(+), 107 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-4.0' into cassandra-4.1
Posted by jl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jlewandowski pushed a commit to branch cassandra-4.1
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 14936d0bd3716ed251e799a264f5ab16d51b893b
Merge: 780f8b94ff 4b9c18235a
Author: Jacek Lewandowski <le...@gmail.com>
AuthorDate: Fri Mar 31 17:30:20 2023 +0200
Merge branch 'cassandra-4.0' into cassandra-4.1
* cassandra-4.0:
Save host id to system.local and flush immediately after startup
CHANGES.txt | 5 +-
.../org/apache/cassandra/db/SystemKeyspace.java | 23 ++-
.../apache/cassandra/db/commitlog/CommitLog.java | 47 ++++--
.../cassandra/db/compaction/CompactionManager.java | 76 +++++++--
.../apache/cassandra/service/StorageService.java | 173 ++++++++++++++++-----
.../cassandra/tools/SSTableMetadataViewer.java | 31 ++--
.../cassandra/distributed/impl/Instance.java | 12 +-
.../cassandra/distributed/impl/InstanceConfig.java | 4 +-
.../distributed/test/IPMembershipTest.java | 4 +
.../distributed/test/SSTableIdGenerationTest.java | 24 +--
.../unit/org/apache/cassandra/db/KeyCacheTest.java | 2 -
.../cassandra/db/compaction/NeverPurgeTest.java | 8 +-
12 files changed, 302 insertions(+), 107 deletions(-)
diff --cc CHANGES.txt
index dec7f680f2,675d423080..c12983568e
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,21 -1,4 +1,24 @@@
-4.0.9
+4.1.2
+ * Return snapshots with dots in their name in nodetool listsnapshots (CASSANDRA-18371)
+ * Fix NPE when loading snapshots and data directory is one directory from root (CASSANDRA-18359)
+ * Do not submit hints when hinted_handoff_enabled=false (CASSANDRA-18304)
+ * Fix COPY ... TO STDOUT behavior in cqlsh (CASSANDRA-18353)
+ * Remove six and Py2SaferScanner merge cruft (CASSANDRA-18354)
-
++Merged from 4.0:
++Merged from 3.11:
++Merged from 3.0:
++ * Save host id to system.local and flush immediately after startup (CASSANDRA-18153)
+
+4.1.1
+ * Deprecate org.apache.cassandra.hadoop code (CASSANDRA-16984)
+ * Fix too early schema version change in sysem local table (CASSANDRA-18291)
+ * Fix copying of JAR of a trigger to temporary file (CASSANDRA-18264)
+ * Fix possible NoSuchFileException when removing a snapshot (CASSANDRA-18211)
+ * PaxosPrepare may add instances to the Electorate that are not in gossip (CASSANDRA-18194)
+ * Fix PAXOS2_COMMIT_AND_PREPARE_RSP serialisation AssertionError (CASSANDRA-18164)
+ * Streaming progress virtual table lock contention can trigger TCP_USER_TIMEOUT and fail streaming (CASSANDRA-18110)
+ * Fix perpetual load of denylist on read in cases where denylist can never be loaded (CASSANDRA-18116)
+Merged from 4.0:
* Fix BufferPool incorrect memoryInUse when putUnusedPortion is used (CASSANDRA-18311)
* Improve memtable allocator accounting when updating AtomicBTreePartition (CASSANDRA-18125)
* Update zstd-jni to version 1.5.4-1 (CASSANDRA-18259)
diff --cc src/java/org/apache/cassandra/db/SystemKeyspace.java
index 1523720d53,d63ee77736..fd2145b30c
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@@ -559,7 -498,18 +561,13 @@@ public final class SystemKeyspac
DECOMMISSIONED
}
- public static void finishStartup()
- {
- Schema.instance.saveSystemKeyspace();
- }
-
public static void persistLocalMetadata()
+ {
+ persistLocalMetadata(UUID::randomUUID);
+ }
+
+ @VisibleForTesting
+ public static void persistLocalMetadata(Supplier<UUID> nodeIdSupplier)
{
String req = "INSERT INTO system.%s (" +
"key," +
diff --cc src/java/org/apache/cassandra/db/commitlog/CommitLog.java
index a832b5ea42,49eb67b1df..6195b1b4ca
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
@@@ -17,19 -17,23 +17,25 @@@
*/
package org.apache.cassandra.db.commitlog;
-
-import java.io.File;
-import java.io.FilenameFilter;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.file.FileStore;
- import java.util.*;
+ import java.util.ArrayList;
+ import java.util.Arrays;
+ import java.util.Collection;
+ import java.util.Collections;
+ import java.util.Iterator;
+ import java.util.List;
+ import java.util.Map;
+ import java.util.TreeMap;
+ import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiPredicate;
import java.util.function.Function;
import java.util.zip.CRC32;
import com.google.common.annotations.VisibleForTesting;
- import org.apache.cassandra.io.util.File;
+import com.google.common.base.Preconditions;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -43,7 -47,7 +49,8 @@@ import org.apache.cassandra.io.compress
import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
-import org.apache.cassandra.io.util.FileUtils;
++import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.PathUtils;
import org.apache.cassandra.metrics.CommitLogMetrics;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.CompressionParams;
@@@ -52,10 -56,9 +59,9 @@@ import org.apache.cassandra.security.En
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MBeanWrapper;
+import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation;
--import static org.apache.cassandra.db.commitlog.CommitLogSegment.CommitLogSegmentFileComparator;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.ENTRY_OVERHEAD_SIZE;
import static org.apache.cassandra.utils.FBUtilities.updateChecksum;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@@@ -70,6 -73,8 +76,8 @@@ public class CommitLog implements Commi
public static final CommitLog instance = CommitLog.construct();
- private static final FilenameFilter unmanagedFilesFilter = (dir, name) -> CommitLogDescriptor.isValid(name) && CommitLogSegment.shouldReplay(name);
++ private static final BiPredicate<File, String> unmanagedFilesFilter = (dir, name) -> CommitLogDescriptor.isValid(name) && CommitLogSegment.shouldReplay(name);
+
final public AbstractCommitLogSegmentManager segmentManager;
public final CommitLogArchiver archiver;
@@@ -82,7 -87,7 +90,6 @@@
private static CommitLog construct()
{
CommitLog log = new CommitLog(CommitLogArchiver.construct(), DatabaseDescriptor.getCommitLogSegmentMgrProvider());
--
MBeanWrapper.instance.registerMBean(log, "org.apache.cassandra.db:type=Commitlog");
return log;
}
@@@ -145,6 -150,24 +152,24 @@@
return this;
}
+ public boolean isStarted()
+ {
+ return started;
+ }
+
+ public boolean hasFilesToReplay()
+ {
+ return getUnmanagedFiles().length > 0;
+ }
+
+ private File[] getUnmanagedFiles()
+ {
- File[] files = new File(segmentManager.storageDirectory).listFiles(unmanagedFilesFilter);
++ File[] files = new File(segmentManager.storageDirectory).tryList(unmanagedFilesFilter);
+ if (files == null)
+ return new File[0];
+ return files;
+ }
+
/**
* Perform recovery on commit logs located in the directory specified by the config file.
*
@@@ -156,12 -179,10 +181,10 @@@
// submit all files for this segment manager for archiving prior to recovery - CASSANDRA-6904
// The files may have already been archived by normal CommitLog operation. This may cause errors in this
// archiving pass, which we should not treat as serious.
- for (File file : new File(segmentManager.storageDirectory).tryList(unmanagedFilesFilter))
+ for (File file : getUnmanagedFiles())
{
- archiver.maybeArchive(file.getPath(), file.getName());
- archiver.maybeWaitForArchiving(file.getName());
+ archiver.maybeArchive(file.path(), file.name());
+ archiver.maybeWaitForArchiving(file.name());
}
assert archiver.archivePending.isEmpty() : "Not all commit log archive tasks were completed before restore";
@@@ -176,7 -197,7 +199,7 @@@
}
else
{
-- Arrays.sort(files, new CommitLogSegmentFileComparator());
++ Arrays.sort(files, new CommitLogSegment.CommitLogSegmentFileComparator());
logger.info("Replaying {}", StringUtils.join(files, ", "));
replayed = recoverFiles(files);
logger.info("Log replay complete, {} replayed mutations", replayed);
diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index bb2d89cda7,b53f3319fa..347f9b3ed0
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@@ -17,8 -17,20 +17,17 @@@
*/
package org.apache.cassandra.db.compaction;
-import java.io.File;
import java.io.IOException;
- import java.util.*;
+ import java.util.ArrayList;
+ import java.util.Arrays;
+ import java.util.Collection;
+ import java.util.Collections;
+ import java.util.HashMap;
+ import java.util.HashSet;
+ import java.util.Iterator;
+ import java.util.List;
+ import java.util.Map;
+ import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@@ -34,25 -48,36 +43,31 @@@ import javax.management.openmbean.Tabul
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
- import com.google.common.collect.*;
+ import com.google.common.collect.ArrayListMultimap;
+ import com.google.common.collect.Collections2;
+ import com.google.common.collect.ConcurrentHashMultiset;
+ import com.google.common.collect.Iterables;
+ import com.google.common.collect.Lists;
+ import com.google.common.collect.Maps;
+ import com.google.common.collect.Multimap;
+ import com.google.common.collect.Multiset;
+ import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListenableFutureTask;
import com.google.common.util.concurrent.RateLimiter;
import com.google.common.util.concurrent.Uninterruptibles;
-
- import org.apache.cassandra.concurrent.ExecutorFactory;
- import org.apache.cassandra.concurrent.WrappedExecutorPlus;
- import org.apache.cassandra.dht.AbstractBounds;
- import org.apache.cassandra.io.util.File;
- import org.apache.cassandra.locator.RangesAtEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.cache.AutoSavingCache;
- import org.apache.cassandra.exceptions.ConfigurationException;
- import org.apache.cassandra.repair.NoSuchRepairSessionException;
- import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
-import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
-import org.apache.cassandra.concurrent.NamedThreadFactory;
++import org.apache.cassandra.concurrent.ExecutorFactory;
++import org.apache.cassandra.concurrent.WrappedExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
- import org.apache.cassandra.schema.Schema;
- import org.apache.cassandra.db.*;
+ import org.apache.cassandra.db.ColumnFamilyStore;
+ import org.apache.cassandra.db.DecoratedKey;
+ import org.apache.cassandra.db.Directories;
+ import org.apache.cassandra.db.DiskBoundaries;
+ import org.apache.cassandra.db.Keyspace;
+ import org.apache.cassandra.db.SerializationHeader;
+ import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.compaction.CompactionInfo.Holder;
import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
@@@ -65,8 -91,8 +81,9 @@@ import org.apache.cassandra.dht.Abstrac
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.index.SecondaryIndexBuilder;
+import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.IndexSummaryRedistribution;
@@@ -75,16 -102,22 +92,26 @@@ import org.apache.cassandra.io.sstable.
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
++import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
+ import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.metrics.CompactionMetrics;
import org.apache.cassandra.metrics.TableMetrics;
++import org.apache.cassandra.repair.NoSuchRepairSessionException;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
+ import org.apache.cassandra.schema.Schema;
+ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.PreviewKind;
- import org.apache.cassandra.utils.*;
+ import org.apache.cassandra.utils.FBUtilities;
+ import org.apache.cassandra.utils.JVMStabilityInspector;
+ import org.apache.cassandra.utils.MBeanWrapper;
+ import org.apache.cassandra.utils.Throwables;
-import org.apache.cassandra.utils.UUIDGen;
++import org.apache.cassandra.utils.TimeUUID;
+ import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.concurrent.Refs;
import static java.util.Collections.singleton;
@@@ -233,6 -259,26 +260,27 @@@ public class CompactionManager implemen
return false;
}
+ @VisibleForTesting
+ public boolean hasOngoingOrPendingTasks()
+ {
+ if (!active.getCompactions().isEmpty() || !compactingCF.isEmpty())
+ return true;
+
+ int pendingTasks = executor.getPendingTaskCount() +
+ validationExecutor.getPendingTaskCount() +
+ viewBuildExecutor.getPendingTaskCount() +
+ cacheCleanupExecutor.getPendingTaskCount();
+ if (pendingTasks > 0)
+ return true;
+
+ int activeTasks = executor.getActiveTaskCount() +
+ validationExecutor.getActiveTaskCount() +
+ viewBuildExecutor.getActiveTaskCount() +
+ cacheCleanupExecutor.getActiveTaskCount();
+
+ return activeTasks > 0;
+ }
++
/**
* Shutdowns both compaction and validation executors, cancels running compaction / validation,
* and waits for tasks to complete if tasks were not cancelable.
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index 6d0d24eb94,e7c6fabc75..92c31c1f7a
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -23,12 -22,29 +23,29 @@@ import java.io.DataInputStream
import java.io.IOError;
import java.io.IOException;
import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
-import java.nio.file.Paths;
+import java.time.Instant;
- import java.util.*;
+ import java.util.ArrayList;
+ import java.util.Arrays;
+ import java.util.Collection;
+ import java.util.Collections;
+ import java.util.EnumMap;
+ import java.util.HashMap;
+ import java.util.HashSet;
+ import java.util.Iterator;
+ import java.util.LinkedHashMap;
+ import java.util.List;
+ import java.util.Map;
import java.util.Map.Entry;
+ import java.util.Optional;
+ import java.util.Scanner;
+ import java.util.Set;
+ import java.util.SortedMap;
+ import java.util.TreeMap;
+ import java.util.TreeSet;
+ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
@@@ -56,44 -71,46 +72,48 @@@ import com.google.common.base.Joiner
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
- import com.google.common.collect.*;
- import com.google.common.util.concurrent.*;
-
- import org.apache.cassandra.config.CassandraRelevantProperties;
- import org.apache.cassandra.concurrent.*;
- import org.apache.cassandra.config.DataStorageSpec;
- import org.apache.cassandra.cql3.QueryHandler;
- import org.apache.cassandra.dht.RangeStreamer.FetchReplica;
- import org.apache.cassandra.fql.FullQueryLogger;
- import org.apache.cassandra.fql.FullQueryLoggerOptions;
- import org.apache.cassandra.fql.FullQueryLoggerOptionsCompositeData;
- import org.apache.cassandra.io.util.File;
- import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
- import org.apache.cassandra.schema.Keyspaces;
- import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster;
- import org.apache.cassandra.utils.concurrent.Future;
- import org.apache.cassandra.schema.TableId;
- import org.apache.cassandra.utils.concurrent.FutureCombiner;
- import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+ import com.google.common.collect.HashMultimap;
+ import com.google.common.collect.ImmutableList;
++import com.google.common.collect.ImmutableMap;
+ import com.google.common.collect.Iterables;
+ import com.google.common.collect.Lists;
+ import com.google.common.collect.Maps;
+ import com.google.common.collect.Multimap;
+ import com.google.common.collect.Ordering;
+ import com.google.common.collect.Sets;
+ import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
+ import com.google.common.util.concurrent.RateLimiter;
+ import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.audit.AuditLogManager;
import org.apache.cassandra.audit.AuditLogOptions;
+import org.apache.cassandra.auth.AuthCacheService;
- import org.apache.cassandra.config.Config.PaxosStatePurging;
- import org.apache.cassandra.service.paxos.*;
- import org.apache.cassandra.service.paxos.cleanup.*;
- import org.apache.cassandra.utils.progress.ProgressListener;
import org.apache.cassandra.auth.AuthKeyspace;
import org.apache.cassandra.auth.AuthSchemaChangeListener;
import org.apache.cassandra.batchlog.BatchlogManager;
+ import org.apache.cassandra.concurrent.ExecutorLocals;
++import org.apache.cassandra.concurrent.FutureTask;
++import org.apache.cassandra.concurrent.FutureTaskWithResources;
+ import org.apache.cassandra.concurrent.NamedThreadFactory;
+ import org.apache.cassandra.concurrent.ScheduledExecutors;
+ import org.apache.cassandra.concurrent.Stage;
++import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config;
++import org.apache.cassandra.config.Config.PaxosStatePurging;
++import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.config.DurationSpec;
+ import org.apache.cassandra.cql3.QueryHandler;
import org.apache.cassandra.cql3.QueryProcessor;
- import org.apache.cassandra.db.*;
+ import org.apache.cassandra.db.ColumnFamilyStore;
+ import org.apache.cassandra.db.DecoratedKey;
-import org.apache.cassandra.db.Directories;
+ import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.Mutation;
+ import org.apache.cassandra.db.SizeEstimatesRecorder;
+ import org.apache.cassandra.db.SnapshotDetailsTabularData;
+ import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.Verifier;
@@@ -108,33 -142,58 +145,74 @@@ import org.apache.cassandra.hints.Hints
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.File;
import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.PathUtils;
- import org.apache.cassandra.locator.*;
+ import org.apache.cassandra.locator.AbstractReplicationStrategy;
+ import org.apache.cassandra.locator.DynamicEndpointSnitch;
+ import org.apache.cassandra.locator.EndpointsByRange;
+ import org.apache.cassandra.locator.EndpointsByReplica;
+ import org.apache.cassandra.locator.EndpointsForRange;
+ import org.apache.cassandra.locator.EndpointsForToken;
+ import org.apache.cassandra.locator.IEndpointSnitch;
+ import org.apache.cassandra.locator.InetAddressAndPort;
+ import org.apache.cassandra.locator.LocalStrategy;
+ import org.apache.cassandra.locator.NetworkTopologyStrategy;
+ import org.apache.cassandra.locator.RangesAtEndpoint;
+ import org.apache.cassandra.locator.RangesByEndpoint;
+ import org.apache.cassandra.locator.Replica;
+ import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
+ import org.apache.cassandra.locator.Replicas;
+ import org.apache.cassandra.locator.SystemReplicas;
+ import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.StorageMetrics;
- import org.apache.cassandra.net.*;
- import org.apache.cassandra.repair.*;
+ import org.apache.cassandra.net.AsyncOneResponse;
+ import org.apache.cassandra.net.Message;
+ import org.apache.cassandra.net.MessagingService;
+ import org.apache.cassandra.repair.RepairRunnable;
-import org.apache.cassandra.repair.SystemDistributedKeyspace;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
import org.apache.cassandra.schema.KeyspaceMetadata;
-import org.apache.cassandra.schema.MigrationCoordinator;
-import org.apache.cassandra.schema.MigrationManager;
++import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.SchemaTransformations;
+import org.apache.cassandra.schema.SystemDistributedKeyspace;
++import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.schema.ViewMetadata;
++import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster;
++import org.apache.cassandra.service.paxos.Paxos;
++import org.apache.cassandra.service.paxos.PaxosCommit;
++import org.apache.cassandra.service.paxos.PaxosRepair;
++import org.apache.cassandra.service.paxos.PaxosState;
++import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupLocalCoordinator;
++import org.apache.cassandra.service.paxos.cleanup.PaxosTableRepairs;
+import org.apache.cassandra.service.snapshot.SnapshotManager;
+import org.apache.cassandra.service.snapshot.TableSnapshot;
- import org.apache.cassandra.net.AsyncOneResponse;
- import org.apache.cassandra.net.MessagingService;
- import org.apache.cassandra.streaming.*;
+ import org.apache.cassandra.streaming.StreamManager;
+ import org.apache.cassandra.streaming.StreamOperation;
+ import org.apache.cassandra.streaming.StreamPlan;
+ import org.apache.cassandra.streaming.StreamResultFuture;
+ import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.tracing.TraceKeyspace;
import org.apache.cassandra.transport.ClientResourceLimits;
import org.apache.cassandra.transport.ProtocolVersion;
- import org.apache.cassandra.utils.*;
++import org.apache.cassandra.utils.ExecutorUtils;
+ import org.apache.cassandra.utils.FBUtilities;
+ import org.apache.cassandra.utils.JVMStabilityInspector;
+ import org.apache.cassandra.utils.MBeanWrapper;
+ import org.apache.cassandra.utils.MD5Digest;
+ import org.apache.cassandra.utils.OutputHandler;
+ import org.apache.cassandra.utils.Pair;
+ import org.apache.cassandra.utils.Throwables;
-import org.apache.cassandra.utils.WindowsTimer;
+ import org.apache.cassandra.utils.WrappedRunnable;
++import org.apache.cassandra.utils.concurrent.Future;
++import org.apache.cassandra.utils.concurrent.FutureCombiner;
++import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.logging.LoggingSupportFactory;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
@@@ -145,10 -205,9 +224,10 @@@ import static com.google.common.collect
import static com.google.common.collect.Iterables.tryFind;
import static java.util.Arrays.asList;
import static java.util.Arrays.stream;
- import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+ import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
@@@ -159,11 -218,8 +238,12 @@@ import static org.apache.cassandra.inde
import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily;
import static org.apache.cassandra.net.NoPayload.noPayload;
import static org.apache.cassandra.net.Verb.REPLICATION_DONE_REQ;
- import static org.apache.cassandra.service.ActiveRepairService.*;
-import static org.apache.cassandra.schema.MigrationManager.evolveSystemKeyspace;
++import static org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
++import static org.apache.cassandra.service.ActiveRepairService.repairCommandExecutor;
+import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
+import static org.apache.cassandra.utils.Clock.Global.nanoTime;
- import static org.apache.cassandra.utils.FBUtilities.now;
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
++import static org.apache.cassandra.utils.FBUtilities.now;
/**
* This abstraction contains the token/identifier of this node
@@@ -2386,7 -2309,15 +2467,15 @@@ public class StorageService extends Not
public UUID getLocalHostUUID()
{
- return getTokenMetadata().getHostId(FBUtilities.getBroadcastAddressAndPort());
+ UUID id = getTokenMetadata().getHostId(FBUtilities.getBroadcastAddressAndPort());
+ if (id != null)
+ return id;
+ // this condition is to prevent accessing the tables when the node is not started yet, and in particular,
+ // when it is not going to be started at all (e.g. when running some unit tests or client tools).
- else if (CommitLog.instance.isStarted())
++ else if ((DatabaseDescriptor.isDaemonInitialized() || DatabaseDescriptor.isToolInitialized()) && CommitLog.instance.isStarted())
+ return SystemKeyspace.getLocalHostId();
+
+ return null;
}
public Map<String, String> getHostIdMap()
diff --cc src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java
index b7164e8d7e,bad7b13150..c4bef35bfd
mode 100644,100755..100644
--- a/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java
+++ b/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java
@@@ -17,14 -17,8 +17,7 @@@
*/
package org.apache.cassandra.tools;
- import static org.apache.cassandra.tools.Util.BLUE;
- import static org.apache.cassandra.tools.Util.CYAN;
- import static org.apache.cassandra.tools.Util.RESET;
- import static org.apache.cassandra.tools.Util.WHITE;
- import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
- import static org.apache.commons.lang3.time.DurationFormatUtils.formatDurationWords;
-
import java.io.DataInputStream;
-import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.io.PrintWriter;
@@@ -65,15 -67,12 +67,13 @@@ import org.apache.cassandra.schema.Tabl
import org.apache.cassandra.tools.Util.TermHistogram;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
- import org.apache.commons.cli.CommandLine;
- import org.apache.commons.cli.CommandLineParser;
- import org.apache.commons.cli.HelpFormatter;
- import org.apache.commons.cli.Option;
- import org.apache.commons.cli.Options;
- import org.apache.commons.cli.ParseException;
- import org.apache.commons.cli.PosixParser;
- import com.google.common.collect.MinMaxPriorityQueue;
+ import static org.apache.cassandra.tools.Util.BLUE;
+ import static org.apache.cassandra.tools.Util.CYAN;
+ import static org.apache.cassandra.tools.Util.RESET;
+ import static org.apache.cassandra.tools.Util.WHITE;
++import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
+ import static org.apache.commons.lang3.time.DurationFormatUtils.formatDurationWords;
/**
* Shows the contents of sstable metadata
@@@ -372,9 -371,10 +372,10 @@@ public class SSTableMetadataViewe
field("maxClusteringValues", Arrays.toString(maxValues));
}
field("Estimated droppable tombstones",
- stats.getEstimatedDroppableTombstoneRatio((int) (System.currentTimeMillis() / 1000) - this.gc));
+ stats.getEstimatedDroppableTombstoneRatio((int) (currentTimeMillis() / 1000) - this.gc));
field("SSTable Level", stats.sstableLevel);
field("Repaired at", stats.repairedAt, toDateString(stats.repairedAt, TimeUnit.MILLISECONDS));
+ field("Originating host id", stats.originatingHostId);
field("Pending repair", stats.pendingRepair);
field("Replay positions covered", stats.commitLogIntervals);
field("totalColumnsSet", stats.totalColumnsSet);
diff --cc test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 44c6f8bf99,b8833dd319..d4cb1cb9ed
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@@ -123,15 -117,9 +123,15 @@@ import org.apache.cassandra.service.Pen
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.StorageServiceMBean;
+import org.apache.cassandra.service.paxos.PaxosRepair;
+import org.apache.cassandra.service.paxos.PaxosState;
++import org.apache.cassandra.service.paxos.uncommitted.UncommittedTableData;
+import org.apache.cassandra.service.reads.thresholds.CoordinatorWarnings;
+import org.apache.cassandra.service.snapshot.SnapshotManager;
+import org.apache.cassandra.streaming.StreamManager;
- import org.apache.cassandra.service.paxos.uncommitted.UncommittedTableData;
import org.apache.cassandra.streaming.StreamReceiveTask;
import org.apache.cassandra.streaming.StreamTransferTask;
-import org.apache.cassandra.streaming.async.StreamingInboundHandler;
+import org.apache.cassandra.streaming.async.NettyStreamingChannel;
import org.apache.cassandra.tools.NodeTool;
import org.apache.cassandra.tools.Output;
import org.apache.cassandra.tools.SystemExitException;
@@@ -599,8 -512,8 +599,8 @@@ public class Instance extends IsolatedE
// We need to persist this as soon as possible after startup checks.
// This should be the first write to SystemKeyspace (CASSANDRA-11742)
- SystemKeyspace.persistLocalMetadata();
+ SystemKeyspace.persistLocalMetadata(config::hostId);
- SystemKeyspaceMigrator40.migrate();
+ SystemKeyspaceMigrator41.migrate();
// Same order to populate tokenMetadata for the first time,
// see org.apache.cassandra.service.CassandraDaemon.setup
@@@ -797,6 -754,9 +797,7 @@@
@Override
public Future<Void> shutdown(boolean graceful)
{
- if (!graceful)
- MessagingService.instance().shutdown(1L, MINUTES, false, true);
-
++ inInstancelogger.info("Shutting down instance {} / {}", config.num(), config.broadcastAddress().getHostString());
Future<?> future = async((ExecutorService executor) -> {
Throwable error = null;
@@@ -810,21 -770,12 +811,26 @@@
}
error = parallelRun(error, executor, StorageService.instance::disableAutoCompaction);
+ while (CompactionManager.instance.hasOngoingOrPendingTasks() && !Thread.currentThread().isInterrupted())
+ {
+ inInstancelogger.info("Waiting for compactions to finish");
+ Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+ // trigger init early or else it could try to init and touch a thread pool that got shutdown
+ HintsService hints = HintsService.instance;
+ ThrowingRunnable shutdownHints = () -> {
+ // this is to allow shutdown in the case hints were halted already
+ try
+ {
+ HintsService.instance.shutdownBlocking();
+ }
+ catch (IllegalStateException e)
+ {
+ if (!"HintsService has already been shut down".equals(e.getMessage()))
+ throw e;
+ }
+ };
error = parallelRun(error, executor,
() -> Gossiper.instance.stopShutdownAndWait(1L, MINUTES),
CompactionManager.instance::forceShutdown,
diff --cc test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index a27daac0e6,037c221a37..92c56d6267
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@@ -28,10 -27,11 +28,9 @@@ import java.util.Map
import java.util.TreeMap;
import java.util.UUID;
import java.util.function.Function;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.stream.Collectors;
import com.vdurmont.semver4j.Semver;
-
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.shared.NetworkTopology;
diff --cc test/distributed/org/apache/cassandra/distributed/test/SSTableIdGenerationTest.java
index 8d05f3dc71,0000000000..dd37bd3d60
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/SSTableIdGenerationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/SSTableIdGenerationTest.java
@@@ -1,531 -1,0 +1,531 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.io.FileUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
+import org.apache.cassandra.db.compaction.DateTieredCompactionStrategy;
+import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
+import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
+import org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
+import org.apache.cassandra.io.sstable.UUIDBasedSSTableId;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.metrics.RestorableMeter;
+import org.apache.cassandra.tools.SystemExitException;
+import org.apache.cassandra.utils.TimeUUID;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.data.Offset;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.Util.bulkLoadSSTables;
+import static org.apache.cassandra.Util.getBackups;
+import static org.apache.cassandra.Util.getSSTables;
+import static org.apache.cassandra.Util.getSnapshots;
+import static org.apache.cassandra.Util.relativizePath;
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.apache.cassandra.db.SystemKeyspace.LEGACY_SSTABLE_ACTIVITY;
+import static org.apache.cassandra.db.SystemKeyspace.SSTABLE_ACTIVITY_V2;
+import static org.apache.cassandra.distributed.shared.FutureUtils.waitOn;
+import static org.apache.cassandra.distributed.test.ExecUtil.rethrow;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class SSTableIdGenerationTest extends TestBaseImpl
+{
+ private final static String ENABLE_UUID_FIELD_NAME = "uuid_sstable_identifiers_enabled";
+ private final static String SNAPSHOT_TAG = "test";
+
+ private int v;
+
+ private static SecurityManager originalSecurityManager;
+
+ @BeforeClass
+ public static void beforeClass() throws Throwable
+ {
+ TestBaseImpl.beforeClass();
+
+ originalSecurityManager = System.getSecurityManager();
+ // we prevent system exit and convert it to exception becuase this is one of the expected test outcomes,
+ // and we want to make an assertion on that
+ ClusterUtils.preventSystemExit();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Throwable
+ {
+ System.setSecurityManager(originalSecurityManager);
+ }
+
+ /**
+ * This test verifies that a node with uuid disabled actually creates sstables with sequential ids and
+ * both the current and legacy sstable activity tables are updated.
+ * Then, when enable uuid, we actually create sstables with uuid but keep and can read the old sstables. Also, only
+ * update the current sstable activity table.
+ */
+ @Test
+ public void testRestartWithUUIDEnabled() throws IOException
+ {
+ try (Cluster cluster = init(Cluster.build(1)
+ .withDataDirCount(1)
+ .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false))
+ .start()))
+ {
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
+ createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
+ assertSSTablesCount(cluster.get(1), 2, 0, KEYSPACE, "tbl");
+ verfiySSTableActivity(cluster, true);
+
+ restartNode(cluster, 1, true);
+
+ createSSTables(cluster.get(1), KEYSPACE, "tbl", 3, 4);
- assertSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, "tbl");
++ assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl");
+ verfiySSTableActivity(cluster, false);
+
+ checkRowsNumber(cluster.get(1), KEYSPACE, "tbl", 9);
+ }
+ }
+
+ /**
+ * This test verifies that we should not be able to start a node with uuid disabled when there are uuid sstables
+ */
+ @Test
+ public void testRestartWithUUIDDisabled() throws IOException
+ {
+ try (Cluster cluster = init(Cluster.build(1)
+ .withDataDirCount(1)
+ .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, true))
+ .start()))
+ {
+ cluster.disableAutoCompaction(KEYSPACE);
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
+ createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
+ assertSSTablesCount(cluster.get(1), 0, 2, KEYSPACE, "tbl");
+ verfiySSTableActivity(cluster, false);
+
+ Assertions.assertThatExceptionOfType(RuntimeException.class)
+ .isThrownBy(() -> restartNode(cluster, 1, false))
+ .withCauseInstanceOf(SystemExitException.class);
+ }
+ }
+
+ @Test
+ public final void testCompactionStrategiesWithMixedSSTables() throws Exception
+ {
+ testCompactionStrategiesWithMixedSSTables(SizeTieredCompactionStrategy.class,
+ DateTieredCompactionStrategy.class,
+ TimeWindowCompactionStrategy.class,
+ LeveledCompactionStrategy.class);
+ }
+
+ /**
+ * The purpose of this test is to verify that we can compact using the given strategy the mix of sstables created
+ * with sequential id and with uuid. Then we verify whether the number results matches the number of rows which we
+ * would get by merging data from the initial sstables.
+ */
+ @SafeVarargs
+ private final void testCompactionStrategiesWithMixedSSTables(final Class<? extends AbstractCompactionStrategy>... compactionStrategyClasses) throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(1)
+ .withDataDirCount(1)
+ .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false))
+ .start()))
+ {
+ // create a table and two sstables with sequential id for each strategy, the sstables will contain overlapping partitions
+ for (Class<? extends AbstractCompactionStrategy> compactionStrategyClass : compactionStrategyClasses)
+ {
+ String tableName = "tbl_" + compactionStrategyClass.getSimpleName().toLowerCase();
+ cluster.schemaChange(createTableStmt(KEYSPACE, tableName, compactionStrategyClass));
+
+ createSSTables(cluster.get(1), KEYSPACE, tableName, 1, 2);
+ assertSSTablesCount(cluster.get(1), 2, 0, KEYSPACE, tableName);
+ }
+
+ // restart the node with uuid enabled
+ restartNode(cluster, 1, true);
+
+ // create another two sstables with uuid for each previously created table
+ for (Class<? extends AbstractCompactionStrategy> compactionStrategyClass : compactionStrategyClasses)
+ {
+ String tableName = "tbl_" + compactionStrategyClass.getSimpleName().toLowerCase();
+
+ createSSTables(cluster.get(1), KEYSPACE, tableName, 3, 4);
+
+ // expect to have a mix of sstables with sequential id and uuid
- assertSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, tableName);
++ assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, tableName);
+
+ // after compaction, we expect to have a single sstable with uuid
+ cluster.get(1).forceCompact(KEYSPACE, tableName);
+ assertSSTablesCount(cluster.get(1), 0, 1, KEYSPACE, tableName);
+
+ // verify the number of rows
+ checkRowsNumber(cluster.get(1), KEYSPACE, tableName, 9);
+ }
+ }
+ }
+
+ @Test
+ public void testStreamingToNodeWithUUIDEnabled() throws Exception
+ {
+ testStreaming(true);
+ }
+
+ @Test
+ public void testStreamingToNodeWithUUIDDisabled() throws Exception
+ {
+ testStreaming(false);
+ }
+
+ /**
+ * The purpose of this test case is to verify the scenario when we need to stream mixed UUID and seq sstables to
+ * a node which have: 1) UUID disabled, and 2) UUID enabled; then verify that we can read all the data properly
+ * from that node alone.
+ */
+ private void testStreaming(boolean uuidEnabledOnTargetNode) throws Exception
+ {
+ // start both nodes with uuid disabled
+ try (Cluster cluster = init(Cluster.build(2)
+ .withDataDirCount(1)
+ .withConfig(config -> config.set(ENABLE_UUID_FIELD_NAME, false).with(Feature.NETWORK))
+ .start()))
+ {
+ // create an empty table and shutdown nodes 2, 3
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl", null));
+ waitOn(cluster.get(2).shutdown());
+
+ // create 2 sstables with overlapping partitions on node 1 (with seq ids)
+ createSSTables(cluster.get(1), KEYSPACE, "tbl", 1, 2);
+
+ // restart node 1 with uuid enabled
+ restartNode(cluster, 1, true);
+
+ // create 2 sstables with overlapping partitions on node 1 (with UUID ids)
+ createSSTables(cluster.get(1), KEYSPACE, "tbl", 3, 4);
+
- assertSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, "tbl");
++ assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl");
+
+ // now start node with UUID disabled and perform repair
+ cluster.get(2).config().set(ENABLE_UUID_FIELD_NAME, uuidEnabledOnTargetNode);
+ cluster.get(2).startup();
+
+ assertSSTablesCount(cluster.get(2), 0, 0, KEYSPACE, "tbl");
+
+ // at this point we have sstables with seq and uuid on nodes and no sstables on node
- // when we run repair, we expect streaming all 5 sstables from node 1 to node 2
++ // when we run repair, we expect streaming all 4 sstables from node 1 to node 2
+
+ cluster.get(2).nodetool("repair", KEYSPACE);
+
+ if (uuidEnabledOnTargetNode)
- assertSSTablesCount(cluster.get(2), 0, 5, KEYSPACE, "tbl");
++ assertSSTablesCount(cluster.get(2), 0, 4, KEYSPACE, "tbl");
+ else
- assertSSTablesCount(cluster.get(2), 5, 0, KEYSPACE, "tbl");
++ assertSSTablesCount(cluster.get(2), 4, 0, KEYSPACE, "tbl");
+
+ waitOn(cluster.get(1).shutdown());
+
+ checkRowsNumber(cluster.get(2), KEYSPACE, "tbl", 9);
+ }
+ }
+
+ @Test
+ public void testSnapshot() throws Exception
+ {
+ File tmpDir = new File(Files.createTempDirectory("test"));
+ Set<String> seqOnlyBackupDirs;
+ Set<String> seqAndUUIDBackupDirs;
+ Set<String> uuidOnlyBackupDirs;
+ try (Cluster cluster = init(Cluster.build(1)
+ .withDataDirCount(1)
+ .withConfig(config -> config.with(Feature.NETWORK)
+ .set("incremental_backups", true)
+ .set("snapshot_before_compaction", false)
+ .set("auto_snapshot", false)
+ .set(ENABLE_UUID_FIELD_NAME, false))
+ .start()))
+ {
+ // create the tables
+
+ cluster.schemaChange("CREATE KEYSPACE new_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
+
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_only", null));
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_and_uuid", null));
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_uuid_only", null));
+ cluster.schemaChange(createTableStmt("new_ks", "tbl_seq_only", null));
+ cluster.schemaChange(createTableStmt("new_ks", "tbl_seq_and_uuid", null));
+ cluster.schemaChange(createTableStmt("new_ks", "tbl_uuid_only", null));
+
+ // creating sstables
+ createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_only", 1, 2, 3, 4);
+ createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 1, 2);
+ createSSTables(cluster.get(1), "new_ks", "tbl_seq_only", 5, 6, 7, 8);
+ createSSTables(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 5, 6);
+
+ restartNode(cluster, 1, true);
+
+ createSSTables(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 3, 4);
+ createSSTables(cluster.get(1), KEYSPACE, "tbl_uuid_only", 1, 2, 3, 4);
+ createSSTables(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 7, 8);
+ createSSTables(cluster.get(1), "new_ks", "tbl_uuid_only", 5, 6, 7, 8);
+
+ Set<String> seqOnlySnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_seq_only");
+ Set<String> seqAndUUIDSnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid");
+ Set<String> uuidOnlySnapshotDirs = snapshot(cluster.get(1), KEYSPACE, "tbl_uuid_only");
+
+ seqOnlyBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_seq_only");
+ seqAndUUIDBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid");
+ uuidOnlyBackupDirs = getBackupDirs(cluster.get(1), KEYSPACE, "tbl_uuid_only");
+
+ // at this point, we should have sstables with backups and snapshots for all tables
- assertSSTablesCount(cluster.get(1), 4, 1, KEYSPACE, "tbl_seq_only");
- assertSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, "tbl_seq_and_uuid");
++ assertSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
++ assertSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
+ assertSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
+
- assertBackupSSTablesCount(cluster.get(1), 4, 1, KEYSPACE, "tbl_seq_only");
- assertBackupSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, "tbl_seq_and_uuid");
++ assertBackupSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
++ assertBackupSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
+ assertBackupSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
+
- assertSnapshotSSTablesCount(cluster.get(1), 4, 1, KEYSPACE, "tbl_seq_only");
- assertSnapshotSSTablesCount(cluster.get(1), 2, 3, KEYSPACE, "tbl_seq_and_uuid");
++ assertSnapshotSSTablesCount(cluster.get(1), 4, 0, KEYSPACE, "tbl_seq_only");
++ assertSnapshotSSTablesCount(cluster.get(1), 2, 2, KEYSPACE, "tbl_seq_and_uuid");
+ assertSnapshotSSTablesCount(cluster.get(1), 0, 4, KEYSPACE, "tbl_uuid_only");
+
+ checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_seq_only", 9);
+ checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_seq_and_uuid", 9);
+ checkRowsNumber(cluster.get(1), KEYSPACE, "tbl_uuid_only", 9);
+
+ // truncate the first set of tables
+ truncateAndAssertEmpty(cluster.get(1), KEYSPACE, "tbl_seq_only", "tbl_seq_and_uuid", "tbl_uuid_only");
+
+ restore(cluster.get(1), seqOnlySnapshotDirs, "tbl_seq_only", 9);
+ restore(cluster.get(1), seqAndUUIDSnapshotDirs, "tbl_seq_and_uuid", 9);
+ restore(cluster.get(1), uuidOnlySnapshotDirs, "tbl_uuid_only", 9);
+
+ truncateAndAssertEmpty(cluster.get(1), KEYSPACE, "tbl_seq_only", "tbl_seq_and_uuid", "tbl_uuid_only");
+
+ restore(cluster.get(1), seqOnlyBackupDirs, "tbl_seq_only", 9);
+ restore(cluster.get(1), seqAndUUIDBackupDirs, "tbl_seq_and_uuid", 9);
+ restore(cluster.get(1), uuidOnlyBackupDirs, "tbl_uuid_only", 9);
+
+ ImmutableSet<String> allBackupDirs = ImmutableSet.<String>builder().addAll(seqOnlyBackupDirs).addAll(seqAndUUIDBackupDirs).addAll(uuidOnlyBackupDirs).build();
+ cluster.get(1).runOnInstance(rethrow(() -> allBackupDirs.forEach(dir -> bulkLoadSSTables(new File(dir), "new_ks"))));
+
+ checkRowsNumber(cluster.get(1), "new_ks", "tbl_seq_only", 17);
+ checkRowsNumber(cluster.get(1), "new_ks", "tbl_seq_and_uuid", 17);
+ checkRowsNumber(cluster.get(1), "new_ks", "tbl_uuid_only", 17);
+
+
+ for (String dir : allBackupDirs)
+ {
+ File src = new File(dir);
+ File dest = relativizePath(tmpDir, src, 3);
+ Files.createDirectories(dest.parent().toPath());
+ FileUtils.moveDirectory(src.toJavaIOFile(), dest.toJavaIOFile());
+ }
+ }
+
+ try (Cluster cluster = init(Cluster.build(1)
+ .withDataDirCount(1)
+ .withConfig(config -> config.with(Feature.NETWORK, Feature.NATIVE_PROTOCOL)
+ .set("incremental_backups", true)
+ .set("snapshot_before_compaction", false)
+ .set("auto_snapshot", false)
+ .set(ENABLE_UUID_FIELD_NAME, false))
+ .start()))
+ {
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_only", null));
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_seq_and_uuid", null));
+ cluster.schemaChange(createTableStmt(KEYSPACE, "tbl_uuid_only", null));
+
+ Function<String, String> relativeToTmpDir = d -> relativizePath(tmpDir, new File(d), 3).toString();
+ restore(cluster.get(1), seqOnlyBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_seq_only", 9);
+ restore(cluster.get(1), seqAndUUIDBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_seq_and_uuid", 9);
+ restore(cluster.get(1), uuidOnlyBackupDirs.stream().map(relativeToTmpDir).collect(Collectors.toSet()), "tbl_uuid_only", 9);
+ }
+ }
+
+ private static void restore(IInvokableInstance instance, Set<String> dirs, String targetTableName, int expectedRowsNum)
+ {
+ List<String> failedImports = instance.callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, targetTableName)
+ .importNewSSTables(dirs, false, false, true, true, true, true, true));
+ assertThat(failedImports).isEmpty();
+ checkRowsNumber(instance, KEYSPACE, targetTableName, expectedRowsNum);
+ }
+
+ private static void truncateAndAssertEmpty(IInvokableInstance instance, String ks, String... tableNames)
+ {
+ for (String tableName : tableNames)
+ {
+ instance.executeInternal(format("TRUNCATE %s.%s", ks, tableName));
+ assertSSTablesCount(instance, 0, 0, ks, tableName);
+ checkRowsNumber(instance, ks, tableName, 0);
+ }
+ }
+
+ private static Set<String> snapshot(IInvokableInstance instance, String ks, String tableName)
+ {
+ Set<String> snapshotDirs = instance.callOnInstance(() -> ColumnFamilyStore.getIfExists(ks, tableName)
+ .snapshot(SNAPSHOT_TAG)
+ .getDirectories()
+ .stream()
+ .map(File::toString)
+ .collect(Collectors.toSet()));
+ assertThat(snapshotDirs).isNotEmpty();
+ return snapshotDirs;
+ }
+
+ private static String createTableStmt(String ks, String name, Class<? extends AbstractCompactionStrategy> compactionStrategy)
+ {
+ if (compactionStrategy == null)
+ compactionStrategy = SizeTieredCompactionStrategy.class;
+ return format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) " +
+ "WITH compaction = {'class':'%s', 'enabled':'false'}",
+ ks, name, compactionStrategy.getCanonicalName());
+ }
+
+ private void createSSTables(IInstance instance, String ks, String tableName, int... records)
+ {
+ String insert = format("INSERT INTO %s.%s (pk, ck, v) VALUES (?, ?, ?)", ks, tableName);
+ for (int record : records)
+ {
+ instance.executeInternal(insert, record, record, ++v);
+ instance.executeInternal(insert, record, record + 1, ++v);
+ instance.executeInternal(insert, record + 1, record + 1, ++v);
+ instance.flush(ks);
+ }
+ }
+
+ private static void assertSSTablesCount(Set<Descriptor> descs, String tableName, int expectedSeqGenIds, int expectedUUIDGenIds)
+ {
+ List<String> seqSSTables = descs.stream().filter(desc -> desc.id instanceof SequenceBasedSSTableId).map(Descriptor::baseFilename).sorted().collect(Collectors.toList());
+ List<String> uuidSSTables = descs.stream().filter(desc -> desc.id instanceof UUIDBasedSSTableId).map(Descriptor::baseFilename).sorted().collect(Collectors.toList());
+ assertThat(seqSSTables).describedAs("SSTables of %s with sequence based id", tableName).hasSize(expectedSeqGenIds);
+ assertThat(uuidSSTables).describedAs("SSTables of %s with UUID based id", tableName).hasSize(expectedUUIDGenIds);
+ }
+
+ private static void assertSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
+ {
+ instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(getSSTables(ks, tableName), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
+ }
+
+ private static void assertSnapshotSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
+ {
+ instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(getSnapshots(ks, tableName, SNAPSHOT_TAG), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
+ }
+
+ private static void assertBackupSSTablesCount(IInvokableInstance instance, int expectedSeqGenIds, int expectedUUIDGenIds, String ks, String... tableNames)
+ {
+ instance.runOnInstance(rethrow(() -> Arrays.stream(tableNames).forEach(tableName -> assertSSTablesCount(getBackups(ks, tableName), tableName, expectedSeqGenIds, expectedUUIDGenIds))));
+ }
+
+ private static Set<String> getBackupDirs(IInvokableInstance instance, String ks, String tableName)
+ {
+ return instance.callOnInstance(() -> getBackups(ks, tableName).stream()
+ .map(d -> d.directory)
+ .map(File::toString)
+ .collect(Collectors.toSet()));
+ }
+
+ private static void verfiySSTableActivity(Cluster cluster, boolean expectLegacyTableIsPopulated)
+ {
+ cluster.get(1).runOnInstance(() -> {
+ RestorableMeter meter = new RestorableMeter(15, 120);
+ SequenceBasedSSTableId seqGenId = new SequenceBasedSSTableId(1);
+ SystemKeyspace.persistSSTableReadMeter("ks", "tab", seqGenId, meter);
+ assertThat(SystemKeyspace.getSSTableReadMeter("ks", "tab", seqGenId)).matches(m -> m.fifteenMinuteRate() == meter.fifteenMinuteRate()
+ && m.twoHourRate() == meter.twoHourRate());
+
+ checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, seqGenId.toString(), true);
+ if (expectLegacyTableIsPopulated)
+ checkSSTableActivityRow(LEGACY_SSTABLE_ACTIVITY, seqGenId.generation, true);
+
+ SystemKeyspace.clearSSTableReadMeter("ks", "tab", seqGenId);
+
+ checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, seqGenId.toString(), false);
+ if (expectLegacyTableIsPopulated)
+ checkSSTableActivityRow(LEGACY_SSTABLE_ACTIVITY, seqGenId.generation, false);
+
+ UUIDBasedSSTableId uuidGenId = new UUIDBasedSSTableId(TimeUUID.Generator.nextTimeUUID());
+ SystemKeyspace.persistSSTableReadMeter("ks", "tab", uuidGenId, meter);
+ assertThat(SystemKeyspace.getSSTableReadMeter("ks", "tab", uuidGenId)).matches(m -> m.fifteenMinuteRate() == meter.fifteenMinuteRate()
+ && m.twoHourRate() == meter.twoHourRate());
+
+ checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, uuidGenId.toString(), true);
+
+ SystemKeyspace.clearSSTableReadMeter("ks", "tab", uuidGenId);
+
+ checkSSTableActivityRow(SSTABLE_ACTIVITY_V2, uuidGenId.toString(), false);
+ });
+ }
+
+ private static void checkSSTableActivityRow(String table, Object genId, boolean expectExists)
+ {
+ String tableColName = SSTABLE_ACTIVITY_V2.equals(table) ? "table_name" : "columnfamily_name";
+ String idColName = SSTABLE_ACTIVITY_V2.equals(table) ? "id" : "generation";
+ String cql = "SELECT rate_15m, rate_120m FROM system.%s WHERE keyspace_name=? and %s=? and %s=?";
+ UntypedResultSet results = executeInternal(format(cql, table, tableColName, idColName), "ks", "tab", genId);
+ assertThat(results).isNotNull();
+
+ if (expectExists)
+ {
+ assertThat(results.isEmpty()).isFalse();
+ UntypedResultSet.Row row = results.one();
+ assertThat(row.getDouble("rate_15m")).isEqualTo(15d, Offset.offset(0.001d));
+ assertThat(row.getDouble("rate_120m")).isEqualTo(120d, Offset.offset(0.001d));
+ }
+ else
+ {
+ assertThat(results.isEmpty()).isTrue();
+ }
+ }
+
+ private static void restartNode(Cluster cluster, int node, boolean uuidEnabled)
+ {
+ waitOn(cluster.get(node).shutdown());
+ cluster.get(node).config().set(ENABLE_UUID_FIELD_NAME, uuidEnabled);
+ cluster.get(node).startup();
+ }
+
+ private static void checkRowsNumber(IInstance instance, String ks, String tableName, int expectedNumber)
+ {
+ SimpleQueryResult result = instance.executeInternalWithResult(format("SELECT * FROM %s.%s", ks, tableName));
+ Object[][] rows = result.toObjectArrays();
+ assertThat(rows).withFailMessage("Invalid results for %s.%s - should have %d rows but has %d: \n%s", ks, tableName, expectedNumber,
+ rows.length, result.toString()).hasSize(expectedNumber);
+ }
+}
diff --cc test/unit/org/apache/cassandra/db/KeyCacheTest.java
index c673458d86,445fc67c00..475f1a162f
--- a/test/unit/org/apache/cassandra/db/KeyCacheTest.java
+++ b/test/unit/org/apache/cassandra/db/KeyCacheTest.java
@@@ -350,10 -348,9 +350,9 @@@ public class KeyCacheTes
CacheService.instance.keyCache.loadSaved();
- // Here max time to load cache is negative which means no time left to load cache. So the keyCache size should
+ // Here max time to load cache is zero which means no time left to load cache. So the keyCache size should
// be zero after loadSaved().
assertKeyCacheSize(0, KEYSPACE1, cf);
- assertEquals(0, CacheService.instance.keyCache.size());
}
@Test
diff --cc test/unit/org/apache/cassandra/db/compaction/NeverPurgeTest.java
index 089ede818e,a5388ca037..42537319ff
--- a/test/unit/org/apache/cassandra/db/compaction/NeverPurgeTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/NeverPurgeTest.java
@@@ -20,9 -20,9 +20,10 @@@ package org.apache.cassandra.db.compact
import java.util.Collection;
+ import org.junit.BeforeClass;
import org.junit.Test;
+import org.apache.cassandra.Util;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org