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,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)

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());

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);
+    }
+}