Rewrite RegisterTest to verify serialization version ceiling

Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-19073

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

@@ -18,9 +18,9 @@
 
 package org.apache.cassandra.distributed.test.log;
 
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
+import java.io.IOException;
 import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
 
 import org.junit.Test;
 
@@ -32,23 +32,28 @@
 import org.apache.cassandra.distributed.shared.NetworkTopology;
 import org.apache.cassandra.distributed.shared.WithProperties;
 import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
 import org.apache.cassandra.tcm.MetadataSnapshots;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.membership.Location;
 import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
 import org.apache.cassandra.tcm.membership.NodeVersion;
 import org.apache.cassandra.tcm.sequences.LeaveStreams;
 import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave;
 import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.tcm.transformations.PrepareLeave;
 import org.apache.cassandra.tcm.transformations.Register;
 import org.apache.cassandra.tcm.transformations.SealPeriod;
+import org.apache.cassandra.tcm.transformations.Startup;
 import org.apache.cassandra.tcm.transformations.Unregister;
 import org.apache.cassandra.utils.CassandraVersion;
-import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.TCM_ALLOW_TRANSFORMATIONS_DURING_UPGRADES;
+import static org.junit.Assert.assertEquals;
 
 public class RegisterTest extends TestBaseImpl
 {
@@ -91,49 +96,70 @@ public void testRegistrationIdempotence() throws Throwable
  }
 
  @Test
- public void serializationVersionDisagreementTest() throws Throwable
+ public void serializationVersionCeilingTest() throws Throwable
  {
- try (Cluster cluster = builder().withNodes(2)
+ try (Cluster cluster = builder().withNodes(1)
  .createWithoutStarting();
  WithProperties prop = new WithProperties().set(TCM_ALLOW_TRANSFORMATIONS_DURING_UPGRADES, "true"))
  {
+ final String firstNodeEndpoint = "127.0.0.10";
  cluster.get(1).startup();
  cluster.get(1).runOnInstance(() -> {
  try
  {
  // Register a ghost node with V0 to fake-force V0 serialization. In a real world cluster we will always be upgrading from a smaller version.
- ClusterMetadataService.instance().commit(new Register(new NodeAddresses(InetAddressAndPort.getByName("127.0.0.10")),
+ ClusterMetadataService.instance().commit(new Register(new NodeAddresses(InetAddressAndPort.getByName(firstNodeEndpoint)),
  ClusterMetadata.current().directory.location(ClusterMetadata.current().myNodeId()),
  new NodeVersion(NodeVersion.CURRENT.cassandraVersion, Version.V0)));
- }
- catch (UnknownHostException e)
- {
- throw new RuntimeException(e);
- }
- ClusterMetadataService.instance().commit(SealPeriod.instance);
- });
-
- cluster.get(2).runOnInstance(() -> {
- try
- {
- Field field = NodeVersion.class.getDeclaredField("CURRENT");
- Field modifiers = Field.class.getDeclaredField("modifiers");
-
- field.setAccessible(true);
- modifiers.setAccessible(true);
-
- int newModifiers = field.getModifiers() & ~Modifier.FINAL;
- modifiers.setInt(field, newModifiers);
- field.set(null, new NodeVersion(new CassandraVersion(FBUtilities.getReleaseVersionString()), NodeVersion.CURRENT_METADATA_VERSION));
+ NodeId oldNode = ClusterMetadata.current().directory.peerId(InetAddressAndPort.getByName(firstNodeEndpoint));
+ // Fake an upgrade of this node and assert we continue to serialize so that the one which only
+ // supports V0 can deserialize. In a real cluster it wouldn't happen exactly in this way (here the
+ // min serialization version actually goes backwards from CURRENT to V0 when we upgrade, which would
+ // not happen in a real cluster as we would never register like oldNode, with the current C* version
+ // but an older metadata version
+ CassandraVersion currentVersion = NodeVersion.CURRENT.cassandraVersion;
+ NodeVersion upgraded = new NodeVersion(new CassandraVersion(String.format("%d.%d.%d", currentVersion.major + 1, 0, 0)),
+ NodeVersion.CURRENT_METADATA_VERSION);
+ ClusterMetadata metadata = ClusterMetadata.current();
+ NodeId id = metadata.myNodeId();
+ Startup startup = new Startup(id, metadata.directory.getNodeAddresses(id), upgraded);
+ ClusterMetadataService.instance().commit(startup);
+ // Doesn't matter which specific Transformation we use here, we're testing that the serializer uses
+ // the correct lower bound
+ Transformation t = new Register(NodeAddresses.current(), new Location("DC", "RACK"), NodeVersion.CURRENT);
+ try
+ {
+ assertEquals(ClusterMetadata.current().directory.clusterMinVersion.serializationVersion,
+ Version.V0.asInt());
+ ByteBuffer bytes = t.kind().toVersionedBytes(t);
+ try (DataInputBuffer buf = new DataInputBuffer(bytes, true))
+ {
+ // Because ClusterMetadata.current().directory still contains oldNode we must serialize at
+ // the version it supports
+ assertEquals(Version.V0, Version.fromInt(buf.readUnsignedVInt32()));
+ }
+
+ // If we unregister oldNode, then the ceiling for serialization version will rise
+ ClusterMetadataService.instance().commit(new Unregister(oldNode));
+ assertEquals(ClusterMetadata.current().directory.clusterMinVersion.serializationVersion,
+ NodeVersion.CURRENT_METADATA_VERSION.asInt());
+ bytes = t.kind().toVersionedBytes(t);
+ try (DataInputBuffer buf = new DataInputBuffer(bytes, true))
+ {
+ assertEquals(NodeVersion.CURRENT_METADATA_VERSION, Version.fromInt(buf.readUnsignedVInt32()));
+ }
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
 
  }
- catch (NoSuchFieldException | IllegalAccessException e)
+ catch (UnknownHostException e)
  {
  throw new RuntimeException(e);
  }
  });
-
- cluster.get(2).startup();
  }
  }