This is an automated email from the ASF dual-hosted git repository. marcuse pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit df1a2d4c3db4ff016d03d5403ac68778a71d5759 Merge: 76f8333 98e798f Author: Marcus Eriksson <marc...@apache.org> AuthorDate: Mon Jan 17 09:13:32 2022 +0100 Merge branch 'cassandra-4.0' into trunk CHANGES.txt | 1 + .../cassandra/repair/RepairMessageVerbHandler.java | 1 + .../schema/SystemDistributedKeyspace.java | 2 +- .../cassandra/service/ActiveRepairService.java | 37 ++++- .../distributed/test/ClearSnapshotTest.java | 170 +++++++++++++++++++++ 5 files changed, 206 insertions(+), 5 deletions(-) diff --cc CHANGES.txt index c3d1e81,0896d56..e5a1a8b --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,82 -1,5 +1,83 @@@ -4.0.2 +4.1 + * Prewarm role and credential caches to avoid timeouts at startup (CASSANDRA-16958) + * Make capacity/validity/updateinterval/activeupdate for Auth Caches configurable via nodetool (CASSANDRA-17063) + * Added startup check for read_ahead_kb setting (CASSANDRA-16436) + * Avoid unecessary array allocations and initializations when performing query checks (CASSANDRA-17209) + * Add guardrail for list operations that require read before write (CASSANDRA-17154) + * Migrate thresholds for number of keyspaces and tables to guardrails (CASSANDRA-17195) + * Remove self-reference in SSTableTidier (CASSANDRA-17205) + * Add guardrail for query page size (CASSANDRA-17189) + * Allow column_index_size_in_kb to be configurable through nodetool (CASSANDRA-17121) + * Emit a metric for number of local read and write calls + * Add non-blocking mode for CDC writes (CASSANDRA-17001) + * Add guardrails framework (CASSANDRA-17147) + * Harden resource management on SSTable components to prevent future leaks (CASSANDRA-17174) + * Make nodes more resilient to local unrelated files during startup (CASSANDRA-17082) + * repair prepare message would produce a wrong error message if network timeout happened rather than reply wait timeout (CASSANDRA-16992) + * Log queries that fail on timeout or unavailable errors up to once per minute by default (CASSANDRA-17159) + * Refactor normal/preview/IR repair to standardize repair cleanup and error handling of failed RepairJobs (CASSANDRA-17069) + * Log missing peers in StartupClusterConnectivityChecker (CASSANDRA-17130) + * Introduce separate rate limiting settings for entire SSTable streaming (CASSANDRA-17065) + * Implement Virtual Tables for Auth Caches (CASSANDRA-16914) + * Actively update auth cache in the background (CASSANDRA-16957) + * Add unix time conversion functions (CASSANDRA-17029) + * JVMStabilityInspector.forceHeapSpaceOomMaybe should handle all non-heap OOMs rather than only supporting direct only (CASSANDRA-17128) + * Forbid other Future implementations with checkstyle (CASSANDRA-17055) + * commit log was switched from non-daemon to daemon threads, which causes the JVM to exit in some case as no non-daemon threads are active (CASSANDRA-17085) + * Add a Denylist to block reads and writes on specific partition keys (CASSANDRA-12106) + * v4+ protocol did not clean up client warnings, which caused leaking the state (CASSANDRA-17054) + * Remove duplicate toCQLString in ReadCommand (CASSANDRA-17023) + * Ensure hint window is persistent across restarts of a node (CASSANDRA-14309) + * Allow to GRANT or REVOKE multiple permissions in a single statement (CASSANDRA-17030) + * Allow to grant permission for all tables in a keyspace (CASSANDRA-17027) + * Log time spent writing keys during compaction (CASSANDRA-17037) + * Make nodetool compactionstats and sstable_tasks consistent (CASSANDRA-16976) + * Add metrics and logging around index summary redistribution (CASSANDRA-17036) + * Add configuration options for minimum allowable replication factor and default replication factor (CASSANDRA-14557) + * Expose information about stored hints via a nodetool command and a virtual table (CASSANDRA-14795) + * Add broadcast_rpc_address to system.local (CASSANDRA-11181) + * Add support for type casting in WHERE clause components and in the values of INSERT/UPDATE statements (CASSANDRA-14337) + * add credentials file support to CQLSH (CASSANDRA-16983) + * Skip remaining bytes in the Envelope buffer when a ProtocolException is thrown to avoid double decoding (CASSANDRA-17026) + * Allow reverse iteration of resources during permissions checking (CASSANDRA-17016) + * Add feature to verify correct ownership of attached locations on disk at startup (CASSANDRA-16879) + * Make SSLContext creation pluggable/extensible (CASSANDRA-16666) + * Add soft/hard limits to local reads to protect against reading too much data in a single query (CASSANDRA-16896) + * Avoid token cache invalidation for removing a non-member node (CASSANDRA-15290) + * Allow configuration of consistency levels on auth operations (CASSANDRA-12988) + * Add number of sstables in a compaction to compactionstats output (CASSANDRA-16844) + * Upgrade Caffeine to 2.9.2 (CASSANDRA-15153) + * Allow DELETE and TRUNCATE to work on Virtual Tables if the implementation allows it (CASSANDRA-16806) + * Include SASI components to snapshots (CASSANDRA-15134) + * Fix missed wait latencies in the output of `nodetool tpstats -F` (CASSANDRA-16938) + * Reduce native transport max frame size to 16MB (CASSANDRA-16886) + * Add support for filtering using IN restrictions (CASSANDRA-14344) + * Provide a nodetool command to invalidate auth caches (CASSANDRA-16404) + * Catch read repair timeout exceptions and add metric (CASSANDRA-16880) + * Exclude Jackson 1.x transitive dependency of hadoop* provided dependencies (CASSANDRA-16854) + * Add client warnings and abort to tombstone and coordinator reads which go past a low/high watermark (CASSANDRA-16850) + * Add TTL support to nodetool snapshots (CASSANDRA-16789) + * Allow CommitLogSegmentReader to optionally skip sync marker CRC checks (CASSANDRA-16842) + * allow blocking IPs from updating metrics about traffic (CASSANDRA-16859) + * Request-Based Native Transport Rate-Limiting (CASSANDRA-16663) + * Implement nodetool getauditlog command (CASSANDRA-16725) + * Clean up repair code (CASSANDRA-13720) + * Background schedule to clean up orphaned hints files (CASSANDRA-16815) + * Modify SecondaryIndexManager#indexPartition() to retrieve only columns for which indexes are actually being built (CASSANDRA-16776) + * Batch the token metadata update to improve the speed (CASSANDRA-15291) + * Reduce the log level on "expected" repair exceptions (CASSANDRA-16775) + * Make JMXTimer expose attributes using consistent time unit (CASSANDRA-16760) + * Remove check on gossip status from DynamicEndpointSnitch::updateScores (CASSANDRA-11671) + * Fix AbstractReadQuery::toCQLString not returning valid CQL (CASSANDRA-16510) + * Log when compacting many tombstones (CASSANDRA-16780) + * Display bytes per level in tablestats for LCS tables (CASSANDRA-16799) + * Add isolated flush timer to CommitLogMetrics and ensure writes correspond to single WaitingOnCommit data points (CASSANDRA-16701) + * Add a system property to set hostId if not yet initialized (CASSANDRA-14582) + * GossiperTest.testHasVersion3Nodes didn't take into account trunk version changes, fixed to rely on latest version (CASSANDRA-16651) + * Update JNA library to 5.9.0 and snappy-java to version 1.1.8.4 (CASSANDRA-17040) + +Merged from 4.0: + * Don't block gossip when clearing repair snapshots (CASSANDRA-17168) * Deduplicate warnings for deprecated parameters (changed names) (CASSANDRA-17160) * Update ant-junit to version 1.10.12 (CASSANDRA-17218) * Add droppable tombstone metrics to nodetool tablestats (CASSANDRA-16308) diff --cc src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java index 9f17578,0000000..6206fca mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java @@@ -1,405 -1,0 +1,405 @@@ +/* + * 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.schema; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import com.google.common.base.Joiner; +import com.google.common.collect.Lists; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Sets; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.repair.CommonRange; +import org.apache.cassandra.repair.messages.RepairOption; +import org.apache.cassandra.utils.FBUtilities; + +import static java.lang.String.format; + +import static org.apache.cassandra.utils.ByteBufferUtil.bytes; + +public final class SystemDistributedKeyspace +{ + private SystemDistributedKeyspace() + { + } + + public static final String NAME = "system_distributed"; + + private static final int DEFAULT_RF = CassandraRelevantProperties.SYSTEM_DISTRIBUTED_DEFAULT_RF.getInt(); + private static final Logger logger = LoggerFactory.getLogger(SystemDistributedKeyspace.class); + + /** + * Generation is used as a timestamp for automatic table creation on startup. + * If you make any changes to the tables below, make sure to increment the + * generation and document your change here. + * + * gen 0: original definition in 2.2 + * gen 1: (pre-)add options column to parent_repair_history in 3.0, 3.11 + * gen 2: (pre-)add coordinator_port and participants_v2 columns to repair_history in 3.0, 3.11, 4.0 + * gen 3: gc_grace_seconds raised from 0 to 10 days in CASSANDRA-12954 in 3.11.0 + * gen 4: compression chunk length reduced to 16KiB, memtable_flush_period_in_ms now unset on all tables in 4.0 + * gen 5: add ttl and TWCS to repair_history tables + * gen 6: add denylist table + */ + public static final long GENERATION = 6; + + public static final String REPAIR_HISTORY = "repair_history"; + + public static final String PARENT_REPAIR_HISTORY = "parent_repair_history"; + + public static final String VIEW_BUILD_STATUS = "view_build_status"; + + public static final String PARTITION_DENYLIST_TABLE = "partition_denylist"; + + private static final TableMetadata RepairHistory = + parse(REPAIR_HISTORY, + "Repair history", + "CREATE TABLE %s (" + + "keyspace_name text," + + "columnfamily_name text," + + "id timeuuid," + + "parent_id timeuuid," + + "range_begin text," + + "range_end text," + + "coordinator inet," + + "coordinator_port int," + + "participants set<inet>," + + "participants_v2 set<text>," + + "exception_message text," + + "exception_stacktrace text," + + "status text," + + "started_at timestamp," + + "finished_at timestamp," + + "PRIMARY KEY ((keyspace_name, columnfamily_name), id))") + .defaultTimeToLive((int) TimeUnit.DAYS.toSeconds(30)) + .compaction(CompactionParams.twcs(ImmutableMap.of("compaction_window_unit","DAYS", + "compaction_window_size","1"))) + .build(); + + private static final TableMetadata ParentRepairHistory = + parse(PARENT_REPAIR_HISTORY, + "Repair history", + "CREATE TABLE %s (" + + "parent_id timeuuid," + + "keyspace_name text," + + "columnfamily_names set<text>," + + "started_at timestamp," + + "finished_at timestamp," + + "exception_message text," + + "exception_stacktrace text," + + "requested_ranges set<text>," + + "successful_ranges set<text>," + + "options map<text, text>," + + "PRIMARY KEY (parent_id))") + .defaultTimeToLive((int) TimeUnit.DAYS.toSeconds(30)) + .compaction(CompactionParams.twcs(ImmutableMap.of("compaction_window_unit","DAYS", + "compaction_window_size","1"))) + .build(); + + private static final TableMetadata ViewBuildStatus = + parse(VIEW_BUILD_STATUS, + "Materialized View build status", + "CREATE TABLE %s (" + + "keyspace_name text," + + "view_name text," + + "host_id uuid," + + "status text," + + "PRIMARY KEY ((keyspace_name, view_name), host_id))").build(); + + public static final TableMetadata PartitionDenylistTable = + parse(PARTITION_DENYLIST_TABLE, + "Partition keys which have been denied access", + "CREATE TABLE %s (" + + "ks_name text," + + "table_name text," + + "key blob," + + "PRIMARY KEY ((ks_name, table_name), key))") + .build(); + + private static TableMetadata.Builder parse(String table, String description, String cql) + { + return CreateTableStatement.parse(format(cql, table), SchemaConstants.DISTRIBUTED_KEYSPACE_NAME) + .id(TableId.forSystemTable(SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, table)) + .comment(description); + } + + public static KeyspaceMetadata metadata() + { + return KeyspaceMetadata.create(SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, KeyspaceParams.simple(Math.max(DEFAULT_RF, DatabaseDescriptor.getDefaultKeyspaceRF())), Tables.of(RepairHistory, ParentRepairHistory, ViewBuildStatus, PartitionDenylistTable)); + } + + public static void startParentRepair(UUID parent_id, String keyspaceName, String[] cfnames, RepairOption options) + { + Collection<Range<Token>> ranges = options.getRanges(); + String query = "INSERT INTO %s.%s (parent_id, keyspace_name, columnfamily_names, requested_ranges, started_at, options)"+ + " VALUES (%s, '%s', { '%s' }, { '%s' }, toTimestamp(now()), { %s })"; + String fmtQry = format(query, + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, + PARENT_REPAIR_HISTORY, + parent_id.toString(), + keyspaceName, + Joiner.on("','").join(cfnames), + Joiner.on("','").join(ranges), + toCQLMap(options.asMap(), RepairOption.RANGES_KEY, RepairOption.COLUMNFAMILIES_KEY)); + processSilent(fmtQry); + } + + private static String toCQLMap(Map<String, String> options, String ... ignore) + { + Set<String> toIgnore = Sets.newHashSet(ignore); + StringBuilder map = new StringBuilder(); + boolean first = true; + for (Map.Entry<String, String> entry : options.entrySet()) + { + if (!toIgnore.contains(entry.getKey())) + { + if (!first) + map.append(','); + first = false; + map.append(format("'%s': '%s'", entry.getKey(), entry.getValue())); + } + } + return map.toString(); + } + + public static void failParentRepair(UUID parent_id, Throwable t) + { + String query = "UPDATE %s.%s SET finished_at = toTimestamp(now()), exception_message=?, exception_stacktrace=? WHERE parent_id=%s"; + + StringWriter sw = new StringWriter(); + PrintWriter pw = new PrintWriter(sw); + t.printStackTrace(pw); + String fmtQuery = format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, PARENT_REPAIR_HISTORY, parent_id.toString()); + String message = t.getMessage(); + processSilent(fmtQuery, message != null ? message : "", sw.toString()); + } + + public static void successfulParentRepair(UUID parent_id, Collection<Range<Token>> successfulRanges) + { + String query = "UPDATE %s.%s SET finished_at = toTimestamp(now()), successful_ranges = {'%s'} WHERE parent_id=%s"; + String fmtQuery = format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, PARENT_REPAIR_HISTORY, Joiner.on("','").join(successfulRanges), parent_id.toString()); + processSilent(fmtQuery); + } + + public static void startRepairs(UUID id, UUID parent_id, String keyspaceName, String[] cfnames, CommonRange commonRange) + { + // Don't record repair history if an upgrade is in progress as version 3 nodes generates errors + // due to schema differences + boolean includeNewColumns = !Gossiper.instance.hasMajorVersion3Nodes(); + + InetAddressAndPort coordinator = FBUtilities.getBroadcastAddressAndPort(); + Set<String> participants = Sets.newHashSet(); + Set<String> participants_v2 = Sets.newHashSet(); + + for (InetAddressAndPort endpoint : commonRange.endpoints) + { + participants.add(endpoint.getHostAddress(false)); + participants_v2.add(endpoint.getHostAddressAndPort()); + } + + String query = + "INSERT INTO %s.%s (keyspace_name, columnfamily_name, id, parent_id, range_begin, range_end, coordinator, coordinator_port, participants, participants_v2, status, started_at) " + + "VALUES ( '%s', '%s', %s, %s, '%s', '%s', '%s', %d, { '%s' }, { '%s' }, '%s', toTimestamp(now()))"; + String queryWithoutNewColumns = + "INSERT INTO %s.%s (keyspace_name, columnfamily_name, id, parent_id, range_begin, range_end, coordinator, participants, status, started_at) " + + "VALUES ( '%s', '%s', %s, %s, '%s', '%s', '%s', { '%s' }, '%s', toTimestamp(now()))"; + + for (String cfname : cfnames) + { + for (Range<Token> range : commonRange.ranges) + { + String fmtQry; + if (includeNewColumns) + { + fmtQry = format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, REPAIR_HISTORY, + keyspaceName, + cfname, + id.toString(), + parent_id.toString(), + range.left.toString(), + range.right.toString(), + coordinator.getHostAddress(false), + coordinator.getPort(), + Joiner.on("', '").join(participants), + Joiner.on("', '").join(participants_v2), + RepairState.STARTED.toString()); + } + else + { + fmtQry = format(queryWithoutNewColumns, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, REPAIR_HISTORY, + keyspaceName, + cfname, + id.toString(), + parent_id.toString(), + range.left.toString(), + range.right.toString(), + coordinator.getHostAddress(false), + Joiner.on("', '").join(participants), + RepairState.STARTED.toString()); + } + processSilent(fmtQry); + } + } + } + + public static void failRepairs(UUID id, String keyspaceName, String[] cfnames, Throwable t) + { + for (String cfname : cfnames) + failedRepairJob(id, keyspaceName, cfname, t); + } + + public static void successfulRepairJob(UUID id, String keyspaceName, String cfname) + { + String query = "UPDATE %s.%s SET status = '%s', finished_at = toTimestamp(now()) WHERE keyspace_name = '%s' AND columnfamily_name = '%s' AND id = %s"; + String fmtQuery = format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, REPAIR_HISTORY, + RepairState.SUCCESS.toString(), + keyspaceName, + cfname, + id.toString()); + processSilent(fmtQuery); + } + + public static void failedRepairJob(UUID id, String keyspaceName, String cfname, Throwable t) + { + String query = "UPDATE %s.%s SET status = '%s', finished_at = toTimestamp(now()), exception_message=?, exception_stacktrace=? WHERE keyspace_name = '%s' AND columnfamily_name = '%s' AND id = %s"; + StringWriter sw = new StringWriter(); + PrintWriter pw = new PrintWriter(sw); + t.printStackTrace(pw); + String fmtQry = format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, REPAIR_HISTORY, + RepairState.FAILED.toString(), + keyspaceName, + cfname, + id.toString()); + String message = t.getMessage(); + if (message == null) + message = t.getClass().getName(); + processSilent(fmtQry, message, sw.toString()); + } + + public static void startViewBuild(String keyspace, String view, UUID hostId) + { + String query = "INSERT INTO %s.%s (keyspace_name, view_name, host_id, status) VALUES (?, ?, ?, ?)"; + QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, VIEW_BUILD_STATUS), + ConsistencyLevel.ONE, + Lists.newArrayList(bytes(keyspace), + bytes(view), + bytes(hostId), + bytes(BuildStatus.STARTED.toString()))); + } + + public static void successfulViewBuild(String keyspace, String view, UUID hostId) + { + String query = "UPDATE %s.%s SET status = ? WHERE keyspace_name = ? AND view_name = ? AND host_id = ?"; + QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, VIEW_BUILD_STATUS), + ConsistencyLevel.ONE, + Lists.newArrayList(bytes(BuildStatus.SUCCESS.toString()), + bytes(keyspace), + bytes(view), + bytes(hostId))); + } + + public static Map<UUID, String> viewStatus(String keyspace, String view) + { + String query = "SELECT host_id, status FROM %s.%s WHERE keyspace_name = ? AND view_name = ?"; + UntypedResultSet results; + try + { + results = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, VIEW_BUILD_STATUS), + ConsistencyLevel.ONE, + keyspace, + view); + } + catch (Exception e) + { + return Collections.emptyMap(); + } + + + Map<UUID, String> status = new HashMap<>(); + for (UntypedResultSet.Row row : results) + { + status.put(row.getUUID("host_id"), row.getString("status")); + } + return status; + } + + public static void setViewRemoved(String keyspaceName, String viewName) + { + String buildReq = "DELETE FROM %s.%s WHERE keyspace_name = ? AND view_name = ?"; + QueryProcessor.executeInternal(format(buildReq, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, VIEW_BUILD_STATUS), keyspaceName, viewName); + forceBlockingFlush(VIEW_BUILD_STATUS); + } + + private static void processSilent(String fmtQry, String... values) + { + try + { + List<ByteBuffer> valueList = new ArrayList<>(values.length); + for (String v : values) + { + valueList.add(bytes(v)); + } - QueryProcessor.process(fmtQry, ConsistencyLevel.ONE, valueList); ++ QueryProcessor.process(fmtQry, ConsistencyLevel.ANY, valueList); + } + catch (Throwable t) + { + logger.error("Error executing query "+fmtQry, t); + } + } + + public static void forceBlockingFlush(String table) + { + if (!DatabaseDescriptor.isUnsafeSystem()) + FBUtilities.waitOnFuture(Keyspace.open(SchemaConstants.DISTRIBUTED_KEYSPACE_NAME).getColumnFamilyStore(table).forceFlush()); + } + + private enum RepairState + { + STARTED, SUCCESS, FAILED + } + + private enum BuildStatus + { + UNKNOWN, STARTED, SUCCESS + } +} diff --cc src/java/org/apache/cassandra/service/ActiveRepairService.java index 7d0a290,f2e8b6e..cc72430 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@@ -20,14 -20,11 +20,15 @@@ package org.apache.cassandra.service import java.io.IOException; import java.net.UnknownHostException; import java.util.*; -import java.util.concurrent.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import javax.management.openmbean.CompositeData; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Predicate; + import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; @@@ -36,10 -33,11 +37,9 @@@ import com.google.common.cache.CacheBui import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import com.google.common.collect.Multimap; -import com.google.common.util.concurrent.AbstractFuture; -import com.google.common.util.concurrent.ListeningExecutorService; --import com.google.common.util.concurrent.MoreExecutors; -import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; +import org.apache.cassandra.concurrent.ExecutorPlus; +import org.apache.cassandra.config.Config; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.locator.EndpointsByRange; import org.apache.cassandra.locator.EndpointsForRange; @@@ -99,16 -95,7 +99,17 @@@ import org.apache.cassandra.utils.concu import static com.google.common.collect.Iterables.concat; import static com.google.common.collect.Iterables.transform; +import static java.util.Collections.synchronizedSet; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; +import static org.apache.cassandra.config.Config.RepairCommandPoolFullStrategy.reject; +import static org.apache.cassandra.config.DatabaseDescriptor.*; +import static org.apache.cassandra.net.Message.out; import static org.apache.cassandra.net.Verb.PREPARE_MSG; +import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; +import static org.apache.cassandra.utils.Simulate.With.MONITORS; ++import static org.apache.cassandra.utils.Clock.Global.nanoTime; +import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch; /** * ActiveRepairService is the starting point for manual "active" repairs. @@@ -189,6 -198,11 +190,10 @@@ public class ActiveRepairService implem private final Gossiper gossiper; private final Cache<Integer, Pair<ParentRepairStatus, List<String>>> repairStatusByCmd; - private final DebuggableThreadPoolExecutor clearSnapshotExecutor = DebuggableThreadPoolExecutor.createWithMaximumPoolSize("RepairClearSnapshot", - 1, - 1, - TimeUnit.HOURS); ++ private final ExecutorPlus clearSnapshotExecutor = executorFactory().configurePooled("RepairClearSnapshot", 1) ++ .withKeepAlive(1, TimeUnit.HOURS) ++ .build(); + public ActiveRepairService(IFailureDetector failureDetector, Gossiper gossiper) { this.failureDetector = failureDetector; @@@ -698,10 -706,22 +703,22 @@@ ParentRepairSession session = parentRepairSessions.remove(parentSessionId); if (session == null) return null; - for (ColumnFamilyStore cfs : session.columnFamilyStores.values()) + + if (session.hasSnapshots) { - if (cfs.snapshotExists(snapshotName)) - cfs.clearSnapshot(snapshotName); + clearSnapshotExecutor.submit(() -> { + logger.info("[repair #{}] Clearing snapshots for {}", parentSessionId, + session.columnFamilyStores.values() + .stream() + .map(cfs -> cfs.metadata().toString()).collect(Collectors.joining(", "))); - long startNanos = System.nanoTime(); ++ long startNanos = nanoTime(); + for (ColumnFamilyStore cfs : session.columnFamilyStores.values()) + { + if (cfs.snapshotExists(snapshotName)) + cfs.clearSnapshot(snapshotName); + } - logger.info("[repair #{}] Cleared snapshots in {}ms", parentSessionId, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos)); ++ logger.info("[repair #{}] Cleared snapshots in {}ms", parentSessionId, TimeUnit.NANOSECONDS.toMillis(nanoTime() - startNanos)); + }); } return session; } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org