From 26c374da4f03e4a6b64e414805cd92f3eb0a36c6 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Thu, 9 Mar 2023 12:11:20 -0800 Subject: [PATCH] Synchronize CQLSSTableWriter#build on the Schema.instance object In this commit the `org.apache.cassandra.io.sstable.CQLSSTableWriter#build` method synchronizes on the `Schema.instance` object (instead of the `CQLSSTableWriter.class`) to prevent concurrent schema operations to fail when the offline tools also updates the schema. For example, a table creation operation, which modifies the keyspace tables metadata, might end up missing the update when a concurrent call to the `CQLSSTableWriter#build` method is accessing the singleton Schema instance. Patch by Francisco Guerrero, reviewed by Yifan Cai, Maxwell Guo, Alex Petrov for CASSANDRA-18317. --- CHANGES.txt | 1 + .../io/sstable/CQLSSTableWriter.java | 2 +- .../CQLSSTableWriterConcurrencyTest.java | 135 ++++++++++++++++++ 3 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java diff --git a/CHANGES.txt b/CHANGES.txt index e06d99b011..25691b1cd9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.12 + * Synchronize CQLSSTableWriter#build on the Schema.instance object (CASSANDRA-18317) * Fix closing iterator in SecondaryIndexBuilder (CASSANDRA-18361) * Update hdrhistogram to 2.1.12 (CASSANDRA-18893) * Improve performance of compactions when table does not have an index (CASSANDRA-18773) diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index 0b8dbae55d..d2c437b00b 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -510,7 +510,7 @@ public class CQLSSTableWriter implements Closeable if (insertStatement == null) throw new IllegalStateException("No insert statement specified, you should provide an insert statement through using()"); - synchronized (CQLSSTableWriter.class) + synchronized (Schema.instance) { if (Schema.instance.getKeyspaceMetadata(SchemaConstants.SCHEMA_KEYSPACE_NAME) == null) Schema.instance.load(Schema.getSystemKeyspaceMetadata()); diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java new file mode 100644 index 0000000000..866a5566b5 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterConcurrencyTest.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.sstable; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Paths; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.schema.Schema; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests modifications to the Schema in the {@link CQLSSTableWriter} class while other Schema modifications are + * occurring concurrently + */ +public class CQLSSTableWriterConcurrencyTest extends CQLTester +{ + private static final Logger LOGGER = LoggerFactory.getLogger(CQLSSTableWriterTest.class); + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + + @BeforeClass + public static void setUpClass() + { + CQLTester.setUpClass(); + } + + @Test + public void testConcurrentSchemaModification() throws InterruptedException, IOException + { + String schema = "CREATE TABLE %s (" + + " k int PRIMARY KEY," + + " v1 text," + + " v2 int" + + ")"; + + int nThreads = 20; + ExecutorService pool = Executors.newFixedThreadPool(nThreads); + CountDownLatch latch = new CountDownLatch(nThreads); + AtomicInteger errorCount = new AtomicInteger(); + + // Prepare all the variables required for the test + String[] tableNames = new String[nThreads]; + String[] fullQueries = new String[nThreads]; + String[] insertStatements = new String[nThreads]; + File[] dataDirs = new File[nThreads]; + String baseDataDir = tempFolder.newFolder().getAbsolutePath(); + + for (int i = 0; i < nThreads; i++) + { + tableNames[i] = String.format("table_%02d", i); + fullQueries[i] = String.format(schema, KEYSPACE + '.' + tableNames[i]); + LOGGER.info(fullQueries[i]); + + if (i % 2 != 0) + { + // dataDir and insert statement are only needed for the CQLSSTableWriter class + dataDirs[i] = Paths.get(baseDataDir, KEYSPACE, tableNames[i]).toFile(); + assert dataDirs[i].mkdirs(); + insertStatements[i] = String.format("INSERT INTO %s.%s (k, v1, v2) VALUES (?, ?, ?)", KEYSPACE, tableNames[i]); + } + + final int finalI = i; + pool.submit(() -> { + try + { + latch.countDown(); + latch.await(); + + // Invoke all schema modifications roughly at the same time + if (finalI % 2 == 0) + { + schemaChange(fullQueries[finalI]); + // If another thread modified the Schema without the proper synchronization, it's possible + // that the table metadata was swapped out and calling the Keyspace#getColumnFamilyStore + // method will produce an IllegalArgumentException + Schema.instance.getKeyspaceInstance(KEYSPACE).getColumnFamilyStore(tableNames[finalI]); + } + else + { + CQLSSTableWriter.builder() + .inDirectory(dataDirs[finalI]) + .forTable(fullQueries[finalI]) + .withPartitioner(Murmur3Partitioner.instance) + .using(insertStatements[finalI]) + .build(); + } + } + catch (Throwable throwable) + { + LOGGER.error("Error while processing element number {}", finalI, throwable); + errorCount.incrementAndGet(); + } + }); + } + + pool.shutdown(); + if (!pool.awaitTermination(1, TimeUnit.MINUTES)) + { + LOGGER.warn("Unable to close executor pool after 1 minute"); + } + assertThat(errorCount.get()).isEqualTo(0); + } +}