Allow empty placements when deserializing cluster metadata

Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-20343

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Allow empty placements when deserializing cluster metadata (CASSANDRA-20343)
  * Reduce heap pressure when initializing CMS (CASSANDRA-20267)
  * Paxos Repair: NoSuchElementException on DistributedSchema.getKeyspaceMetadata (CASSANDRA-20320)
  * Improve performance of DistributedSchema.validate for large schemas (CASSANDRA-20360)

src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java

@@ -487,9 +487,8 @@ public class ReplicaGroups
                     InetAddressAndPort replicaAddress = InetAddressAndPort.MetadataSerializer.serializer.deserialize(in, version);
                     boolean isFull = in.readBoolean();
                     replicas.add(new Replica(replicaAddress, replicaRange, isFull));
-
                 }
-                EndpointsForRange efr = EndpointsForRange.copyOf(replicas);
+                EndpointsForRange efr = replicas.isEmpty() ? EndpointsForRange.builder(range).build() : EndpointsForRange.copyOf(replicas);
                 result.put(range, VersionedEndpoints.forRange(lastModified, efr));
             }
             return new ReplicaGroups(result);

test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataSerializationRoundTripTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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.log;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
+import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+public class ClusterMetadataSerializationRoundTripTest extends TestBaseImpl
+{
+    @Test
+    public void testEmptyPlacements() throws IOException
+    {
+        try (Cluster cluster = init(builder().withNodes(1)
+                                             .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP))
+                                             .withInstanceInitializer(BBHelper::install)
+                                             .start()))
+        {
+            cluster.schemaChange("create keyspace x with replication = { 'class': 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc1': '3'}");
+            cluster.get(1).runOnInstance(() -> {
+                Epoch epoch = ClusterMetadata.current().epoch;
+                try
+                {
+                    String dump = ClusterMetadataService.instance().dumpClusterMetadata(epoch, epoch, NodeVersion.CURRENT_METADATA_VERSION);
+                    ClusterMetadataService.instance().loadClusterMetadata(dump);
+                }
+                catch (IOException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            });
+        }
+    }
+
+    // shame, allow us to create a keyspace with a bad replication strategy
+    public static class BBHelper
+    {
+        public static void install(ClassLoader cl, int i)
+        {
+            new ByteBuddy().rebase(ReplicationParams.class)
+                           .method(named("validate").and(takesArguments(3)))
+                           .intercept(MethodDelegation.to(BBHelper.class))
+                           .make()
+                           .load(cl, ClassLoadingStrategy.Default.INJECTION);
+            new ByteBuddy().rebase(NetworkTopologyStrategy.class)
+                           .method(named("validateExpectedOptions").and(takesArguments(1)))
+                           .intercept(MethodDelegation.to(BBHelper.class))
+                           .make()
+                           .load(cl, ClassLoadingStrategy.Default.INJECTION);
+        }
+
+        public static void validate(String name, ClientState state, ClusterMetadata metadata) {}
+
+        public static void validateExpectedOptions(ClusterMetadata metadata) throws ConfigurationException {}
+    }
+}