Reject empty options and invalid DC names in replication configuratio…

…n while creating or altering a keyspace (CASSANDRA-12681)

src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java

@@ -320,7 +320,7 @@ protected void validateReplicationFactor(String rf) throws ConfigurationExceptio
         }
     }
 
-    private void validateExpectedOptions() throws ConfigurationException
+    protected void validateExpectedOptions() throws ConfigurationException
     {
         Collection expectedOptions = recognizedOptions();
         if (expectedOptions == null)

src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java

@@ -24,9 +24,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.locator.TokenMetadata.Topology;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 
@@ -225,6 +227,45 @@ public void validateOptions() throws ConfigurationException
         }
     }
 
+    /*
+     * (non-javadoc) Method to generate list of valid data center names to be used to validate the replication parameters during CREATE / ALTER keyspace operations.
+     * All peers of current node are fetched from {@link TokenMetadata} and then a set is build by fetching DC name of each peer.
+     * @return a set of valid DC names
+     */
+    private static Set<String> buildValidDataCentersSet()
+    {
+        final Set<String> validDataCenters = new HashSet<>();
+        final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+
+        // Add data center of localhost.
+        validDataCenters.add(snitch.getDatacenter(FBUtilities.getBroadcastAddress()));
+        // Fetch and add DCs of all peers.
+        for (final InetAddress peer : StorageService.instance.getTokenMetadata().getAllEndpoints())
+        {
+            validDataCenters.add(snitch.getDatacenter(peer));
+        }
+
+        return validDataCenters;
+    }
+
+    public Collection<String> recognizedOptions()
+    {
+        // only valid options are valid DC names.
+        return buildValidDataCentersSet();
+    }
+
+    protected void validateExpectedOptions() throws ConfigurationException
+    {
+        // Do not accept query with no data centers specified.
+        if (this.configOptions.isEmpty())
+        {
+            throw new ConfigurationException("Configuration for at least one datacenter must be present");
+        }
+
+        // Validate the data center names
+        super.validateExpectedOptions();
+    }
+
     @Override
     public boolean hasSameSettings(AbstractReplicationStrategy other)
     {

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -58,6 +58,8 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.SyntaxException;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.locator.AbstractEndpointSnitch;
+import org.apache.cassandra.locator.IEndpointSnitch;
 import org.apache.cassandra.serializers.TypeSerializer;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
@@ -87,6 +89,8 @@ public abstract class CQLTester
     protected static final long ROW_CACHE_SIZE_IN_MB = Integer.valueOf(System.getProperty("cassandra.test.row_cache_size_in_mb", "0"));
     private static final AtomicInteger seqNumber = new AtomicInteger();
     protected static final ByteBuffer TOO_BIG = ByteBuffer.allocate(FBUtilities.MAX_UNSIGNED_SHORT + 1024);
+    public static final String DATA_CENTER = "datacenter1";
+    public static final String RACK1 = "rack1";
 
     private static org.apache.cassandra.transport.Server server;
     protected static final int nativePort;
@@ -142,6 +146,13 @@ public static final ProtocolVersion getDefaultVersion()
         {
             throw new RuntimeException(e);
         }
+        // Register an EndpointSnitch which returns fixed values for test.
+        DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
+        {
+            @Override public String getRack(InetAddress endpoint) { return RACK1; }
+            @Override public String getDatacenter(InetAddress endpoint) { return DATA_CENTER; }
+            @Override public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2) { return 0; }
+        });
     }
 
     public static ResultMessage lastSchemaChangeResult;

test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java

@@ -246,16 +246,6 @@ public void testUnknownCompressionOptions() throws Throwable
                                                                        tableName));
     }
 
-    /**
-     * Check one can use arbitrary name for datacenter when creating keyspace (#4278),
-     * migrated from cql_tests.py:TestCQL.keyspace_creation_options_test()
-     */
-    @Test
-    public void testDataCenterName() throws Throwable
-    {
-       execute("CREATE KEYSPACE Foo WITH replication = { 'class' : 'NetworkTopologyStrategy', 'us-east' : 1, 'us-west' : 1 };");
-    }
-
     /**
      * Migrated from cql_tests.py:TestCQL.indexes_composite_test()
      */

test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java

@@ -211,13 +211,13 @@ public void testCreateAlterKeyspaces() throws Throwable
                    row(ks1, true),
                    row(ks2, false));
 
-        schemaChange("ALTER KEYSPACE " + ks1 + " WITH replication = { 'class' : 'NetworkTopologyStrategy', 'dc1' : 1 } AND durable_writes=False");
+        schemaChange("ALTER KEYSPACE " + ks1 + " WITH replication = { 'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 1 } AND durable_writes=False");
         schemaChange("ALTER KEYSPACE " + ks2 + " WITH durable_writes=true");
 
         assertRowsIgnoringOrderAndExtra(execute("SELECT keyspace_name, durable_writes, replication FROM system_schema.keyspaces"),
                    row(KEYSPACE, true, map("class", "org.apache.cassandra.locator.SimpleStrategy", "replication_factor", "1")),
                    row(KEYSPACE_PER_TEST, true, map("class", "org.apache.cassandra.locator.SimpleStrategy", "replication_factor", "1")),
-                   row(ks1, false, map("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", "dc1", "1")),
+                   row(ks1, false, map("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", DATA_CENTER , "1")),
                    row(ks2, true, map("class", "org.apache.cassandra.locator.SimpleStrategy", "replication_factor", "1")));
 
         execute("USE " + ks1);
@@ -237,11 +237,11 @@ public void testAlterKeyspaceWithMultipleInstancesOfSameDCThrowsSyntaxException(
         try
         {
             // Create a keyspace
-            execute("CREATE KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', 'dc1' : 2}");
+            execute("CREATE KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER +"' : 2}");
 
             // try modifying the keyspace
-            assertInvalidThrow(SyntaxException.class, "ALTER KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', 'dc1' : 2, 'dc1' : 3 }");
-            execute("ALTER KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', 'dc1' : 3}");
+            assertInvalidThrow(SyntaxException.class, "ALTER KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', " + DATA_CENTER + " : 2, " + DATA_CENTER + " : 3 }");
+            execute("ALTER KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 3}");
         }
         finally
         {
@@ -250,6 +250,49 @@ public void testAlterKeyspaceWithMultipleInstancesOfSameDCThrowsSyntaxException(
         }
     }
 
+    /**
+     * Test {@link ConfigurationException} thrown on alter keyspace to no DC option in replication configuration.
+     */
+    @Test
+    public void testAlterKeyspaceWithNoOptionThrowsConfigurationException() throws Throwable
+    {
+        // Create keyspaces
+        execute("CREATE KEYSPACE testABC WITH replication={ 'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 3 }");
+        execute("CREATE KEYSPACE testXYZ WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 3 }");
+
+        // Try to alter the created keyspace without any option
+        assertInvalidThrow(ConfigurationException.class, "ALTER KEYSPACE testABC WITH replication={ 'class' : 'NetworkTopologyStrategy' }");
+        assertInvalidThrow(ConfigurationException.class, "ALTER KEYSPACE testXYZ WITH replication={ 'class' : 'SimpleStrategy' }");
+
+        // Make sure that the alter works as expected
+        execute("ALTER KEYSPACE testABC WITH replication={ 'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 2 }");
+        execute("ALTER KEYSPACE testXYZ WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 2 }");
+
+        // clean up
+        execute("DROP KEYSPACE IF EXISTS testABC");
+        execute("DROP KEYSPACE IF EXISTS testXYZ");
+    }
+
+    /**
+     * Test {@link ConfigurationException} thrown when altering a keyspace to invalid DC option in replication configuration.
+     */
+    @Test
+    public void testAlterKeyspaceWithNTSOnlyAcceptsConfiguredDataCenterNames() throws Throwable
+    {
+        // Create a keyspace with expected DC name.
+        execute("CREATE KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 2 }");
+
+        // try modifying the keyspace
+        assertInvalidThrow(ConfigurationException.class, "ALTER KEYSPACE testABC WITH replication = { 'class' : 'NetworkTopologyStrategy', 'INVALID_DC' : 2 }");
+        execute("ALTER KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 3 }");
+
+        // Mix valid and invalid, should throw an exception
+        assertInvalidThrow(ConfigurationException.class, "ALTER KEYSPACE testABC WITH replication={ 'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 2 , 'INVALID_DC': 1}");
+
+        // clean-up
+        execute("DROP KEYSPACE IF EXISTS testABC");
+    }
+
     /**
      * Test for bug of 5232,
      * migrated from cql_tests.py:TestCQL.alter_bug_test()

test/unit/org/apache/cassandra/cql3/validation/operations/CreateTest.java

@@ -19,13 +19,15 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.net.InetAddress;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.UUID;
 
 import org.junit.Test;
 
 import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.config.SchemaConstants;
 import org.apache.cassandra.cql3.CQLTester;
@@ -35,6 +37,8 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.SyntaxException;
 import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.AbstractEndpointSnitch;
+import org.apache.cassandra.locator.IEndpointSnitch;
 import org.apache.cassandra.schema.SchemaKeyspace;
 import org.apache.cassandra.triggers.ITrigger;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -519,6 +523,33 @@ public void testCreateKeyspaceWithMultipleInstancesOfSameDCThrowsException() thr
         }
     }
 
+    /**
+     *  Test {@link ConfigurationException} is thrown on create keyspace with invalid DC option in replication configuration .
+     */
+    @Test
+    public void testCreateKeyspaceWithNTSOnlyAcceptsConfiguredDataCenterNames() throws Throwable
+    {
+        assertInvalidThrow(ConfigurationException.class, "CREATE KEYSPACE testABC WITH replication = { 'class' : 'NetworkTopologyStrategy', 'INVALID_DC' : 2 }");
+        execute("CREATE KEYSPACE testABC WITH replication = {'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 2 }");
+
+        // Mix valid and invalid, should throw an exception
+        assertInvalidThrow(ConfigurationException.class, "CREATE KEYSPACE testXYZ WITH replication={ 'class' : 'NetworkTopologyStrategy', '" + DATA_CENTER + "' : 2 , 'INVALID_DC': 1}");
+
+        // clean-up
+        execute("DROP KEYSPACE IF EXISTS testABC");
+        execute("DROP KEYSPACE IF EXISTS testXYZ");
+    }
+
+    /**
+     * Test {@link ConfigurationException} is thrown on create keyspace without any options.
+     */
+    @Test
+    public void testConfigurationExceptionThrownWhenCreateKeyspaceWithNoOptions() throws Throwable
+    {
+        assertInvalidThrow(ConfigurationException.class, "CREATE KEYSPACE testXYZ with replication = { 'class': 'NetworkTopologyStrategy' }");
+        assertInvalidThrow(ConfigurationException.class, "CREATE KEYSPACE testXYZ WITH replication = { 'class' : 'SimpleStrategy' }");
+    }
+
     /**
      * Test create and drop table
      * migrated from cql_tests.py:TestCQL.table_test()
@@ -682,6 +713,34 @@ public void testCreateIndextWithCompactStaticFormat() throws Throwable
                              "CREATE INDEX value_index on %s (value)");
     }
 
+    @Test
+    // tests CASSANDRA-4278
+    public void testHyphenDatacenters() throws Throwable
+    {
+        IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+
+        // Register an EndpointSnitch which returns fixed values for test.
+        DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
+        {
+            @Override
+            public String getRack(InetAddress endpoint) { return RACK1; }
+
+            @Override
+            public String getDatacenter(InetAddress endpoint) { return "us-east-1"; }
+
+            @Override
+            public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2) { return 0; }
+        });
+
+        execute("CREATE KEYSPACE Foo WITH replication = { 'class' : 'NetworkTopologyStrategy', 'us-east-1' : 1 };");
+
+        // Restore the previous EndpointSnitch
+        DatabaseDescriptor.setEndpointSnitch(snitch);
+
+        // Clean up
+        execute("DROP KEYSPACE IF EXISTS Foo");
+    }
+
     @Test
     // tests CASSANDRA-9565
     public void testDoubleWith() throws Throwable

test/unit/org/apache/cassandra/dht/BootStrapperTest.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 
 import com.google.common.collect.Lists;
 
@@ -59,7 +60,7 @@ public class BootStrapperTest
     static IPartitioner oldPartitioner;
 
     @BeforeClass
-    public static void setup() throws ConfigurationException
+    public static void setup() throws Exception
     {
         DatabaseDescriptor.daemonInitialization();
         oldPartitioner = StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
@@ -178,6 +179,13 @@ public void testAllocateTokensNetworkStrategy(int rackCount, int replicas) throw
             int vn = 16;
             String ks = "BootStrapperTestNTSKeyspace" + rackCount + replicas;
             String dc = "1";
+
+            // Register peers with expected DC for NetworkTopologyStrategy.
+            TokenMetadata metadata = StorageService.instance.getTokenMetadata();
+            metadata.clearUnsafe();
+            metadata.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.1.0.99"));
+            metadata.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.15.0.99"));
+
             SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1"));
             TokenMetadata tm = StorageService.instance.getTokenMetadata();
             tm.clearUnsafe();

test/unit/org/apache/cassandra/service/MoveTest.java

@@ -82,7 +82,7 @@ public class MoveTest
      * So instead of extending SchemaLoader, we call it's method below.
      */
     @BeforeClass
-    public static void setup() throws ConfigurationException
+    public static void setup() throws Exception
     {
         DatabaseDescriptor.daemonInitialization();
         oldPartitioner = StorageService.instance.setPartitionerUnsafe(partitioner);
@@ -106,7 +106,7 @@ public void clearTokenMetadata()
         StorageService.instance.getTokenMetadata().clearUnsafe();
     }
 
-    private static void addNetworkTopologyKeyspace(String keyspaceName, Integer... replicas) throws ConfigurationException
+    private static void addNetworkTopologyKeyspace(String keyspaceName, Integer... replicas) throws Exception
     {
 
         DatabaseDescriptor.setEndpointSnitch(new AbstractNetworkTopologySnitch()
@@ -140,6 +140,11 @@ private int getIPLastPart(InetAddress endpoint)
             }
         });
 
+        final TokenMetadata tmd = StorageService.instance.getTokenMetadata();
+        tmd.clearUnsafe();
+        tmd.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.0.0.1"));
+        tmd.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.0.0.2"));
+
         KeyspaceMetadata keyspace =  KeyspaceMetadata.create(keyspaceName,
                                                              KeyspaceParams.nts(configOptions(replicas)),
                                                              Tables.of(CFMetaData.Builder.create(keyspaceName, "CF1")