Merge branch 'cassandra-4.1' into cassandra-5.0

* cassandra-4.1:
  Fix filtering system ks sstables for relocation on startup
This commit is contained in:
Jacek Lewandowski 2023-11-28 18:49:05 +01:00
commit 3cdf71defe
3 changed files with 91 additions and 7 deletions

View File

@ -16,6 +16,7 @@
* Remove deprecated code in Cassandra 1.x and 2.x (CASSANDRA-18959)
* ClientRequestSize metrics should not treat CONTAINS restrictions as being equality-based (CASSANDRA-18896)
Merged from 4.0:
* Fix filtering system ks sstables for relocation on startup (CASSANDRA-18963)
* Remove completed coordinator sessions (CASSANDRA-18903)

View File

@ -28,21 +28,18 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import javax.management.ObjectName;
import javax.management.StandardMBean;
import javax.management.remote.JMXConnectorServer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistryListener;
import com.codahale.metrics.SharedMetricRegistries;
import org.apache.cassandra.audit.AuditLogManager;
import org.apache.cassandra.auth.AuthCacheService;
import org.apache.cassandra.concurrent.ScheduledExecutors;
@ -70,8 +67,8 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.service.paxos.PaxosState;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -509,9 +506,7 @@ public class CassandraDaemon
try (Stream<Path> keyspaceChildren = Files.list(keyspaceDirectory))
{
Path[] tableDirectories = keyspaceChildren.filter(Files::isDirectory)
.filter(p -> !SystemKeyspace.TABLES_SPLIT_ACROSS_MULTIPLE_DISKS
.contains(p.getFileName()
.toString()))
.filter(p -> SystemKeyspace.TABLES_SPLIT_ACROSS_MULTIPLE_DISKS.stream().noneMatch(t -> p.getFileName().toString().startsWith(t + '-')))
.toArray(Path[]::new);
for (Path tableDirectory : tableDirectories)

View File

@ -0,0 +1,88 @@
/*
* 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.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import org.junit.Test;
import org.apache.cassandra.cql3.QueryHandler;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import static com.datastax.driver.core.ParseUtils.doubleQuote;
import static org.assertj.core.api.Assertions.assertThat;
public class SystemKeyspacesDataLocationTest extends TestBaseImpl
{
@Test
public void preparedStatementsShouldNotBeMovedOnStartupTest() throws IOException, ExecutionException, InterruptedException
{
try (Cluster cluster = Cluster.build(1).withDataDirCount(3).start())
{
// to test the behaviour we need have sstables of prepared statements in more than one data directory
cluster.get(1).runOnInstance(() -> {
ColumnFamilyStore preparedStatementsCFS = ColumnFamilyStore.getIfExists(SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.PREPARED_STATEMENTS);
preparedStatementsCFS.disableAutoCompaction();
createSSTablesForPreparedStatementsTable(preparedStatementsCFS);
});
Set<String> preparedStatementsDataLocationsBefore = getPreparedStatementsDataLocations(cluster.get(1));
assertThat(preparedStatementsDataLocationsBefore).hasSizeGreaterThan(1);
cluster.get(1).shutdown().get();
// we expect that for prepared statements, sstables will not be moved to the first data directory on startup
cluster.get(1).startup();
Set<String> preparedStatementsDataLocationsAfter = getPreparedStatementsDataLocations(cluster.get(1));
assertThat(preparedStatementsDataLocationsAfter).isEqualTo(preparedStatementsDataLocationsBefore);
}
}
private static Set<String> getPreparedStatementsDataLocations(IInvokableInstance i)
{
return i.callOnInstance(() -> ColumnFamilyStore.getIfExists(SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.PREPARED_STATEMENTS)
.getLiveSSTables().stream()
// sstables are located at <data-location>/<keyspace-name>/<table-name-with-id> so we need to go up two directories to get the data location
.map(sstr -> sstr.descriptor.directory.parent().parent().toString())
.collect(Collectors.toSet()));
}
private static void createSSTablesForPreparedStatementsTable(ColumnFamilyStore preparedStatementsCFS)
{
for (String keyspaceName : SchemaConstants.LOCAL_SYSTEM_KEYSPACE_NAMES)
{
for (TableMetadata tableMetadata : Schema.instance.getKeyspaceMetadata(keyspaceName).tables)
{
String query = String.format("select * from %s.%s", keyspaceName, doubleQuote(tableMetadata.name));
QueryHandler.Prepared prepared = QueryProcessor.prepareInternal(query);
QueryProcessor.storePreparedStatement(query, keyspaceName, prepared);
preparedStatementsCFS.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
}
}
}
}