diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/RegisterTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/RegisterTest.java index 81006bbfc0..d12dddeef5 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/RegisterTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/RegisterTest.java @@ -19,12 +19,13 @@ package org.apache.cassandra.distributed.test.log; import java.io.IOException; -import java.net.UnknownHostException; +import java.io.Serializable; import java.nio.ByteBuffer; import java.util.EnumSet; import org.junit.Test; +import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; @@ -33,9 +34,11 @@ import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.shared.NetworkTopology; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.io.util.DataInputBuffer; -import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.MetadataSnapshots; import org.apache.cassandra.tcm.Transformation; import org.apache.cassandra.tcm.membership.Location; @@ -45,15 +48,21 @@ import org.apache.cassandra.tcm.membership.NodeState; import org.apache.cassandra.tcm.membership.NodeVersion; import org.apache.cassandra.tcm.ownership.PlacementProvider; import org.apache.cassandra.tcm.sequences.LeaveStreams; +import org.apache.cassandra.tcm.sequences.LockedRanges; import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave; +import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.tcm.transformations.CustomTransformation; import org.apache.cassandra.tcm.transformations.PrepareLeave; import org.apache.cassandra.tcm.transformations.Register; import org.apache.cassandra.tcm.transformations.TriggerSnapshot; import org.apache.cassandra.tcm.transformations.Unregister; -import org.apache.cassandra.utils.CassandraVersion; +import org.apache.cassandra.utils.vint.VIntCoding; +import static org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper.addr; +import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT_METADATA_VERSION; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; public class RegisterTest extends TestBaseImpl { @@ -106,57 +115,42 @@ public class RegisterTest extends TestBaseImpl { cluster.get(1).startup(); cluster.get(1).runOnInstance(() -> { + + // Run a custom transformation to inject a fake node into the directory with a known id and an + // artificially lowered max supported serialization version + CustomTransformation.registerExtension(RegisterNodeWithOldVersion.NAME, RegisterNodeWithOldVersion.serializer); + CustomTransformation injectOldNode = new CustomTransformation(RegisterNodeWithOldVersion.NAME, + new RegisterNodeWithOldVersion()); + ClusterMetadataService.instance().commit(injectOldNode); + + // 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(), TEST_LOCATION, NodeVersion.CURRENT); try { - // Unregister to make directory empty - ClusterMetadataService.instance().commit(new Unregister(ClusterMetadata.current().myNodeId(), - EnumSet.allOf(NodeState.class), - ClusterMetadataService.instance().placementProvider())); - - // Register a ghost node with V0 (bypasses version check because directory is now empty). - // 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.100")), - TEST_LOCATION, - new NodeVersion(NodeVersion.CURRENT.cassandraVersion, Version.V0))); - NodeId oldNode = ClusterMetadata.current().directory.peerId(InetAddressAndPort.getByName("127.0.0.100")); - - // Register a node with upgraded 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); - ClusterMetadataService.instance().commit(new Register(new NodeAddresses(InetAddressAndPort.getByName("127.0.0.200")), TEST_LOCATION, upgraded)); - - // 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.commonSerializationVersion, RegisterNodeWithOldVersion.METADATA_VERSION); + ByteBuffer bytes = t.kind().toVersionedBytes(t); + try (DataInputBuffer buf = new DataInputBuffer(bytes, true)) { - 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, EnumSet.allOf(NodeState.class), ClusterMetadataService.instance().placementProvider())); - 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())); - } + // Because ClusterMetadata.current().directory still contains the fake old node we must + // serialize at the version _it_ supports + assertEquals(RegisterNodeWithOldVersion.METADATA_VERSION, Version.fromInt(buf.readUnsignedVInt32())); } - catch (IOException e) + + // If we unregister the fake node, then the ceiling for serialization version will rise + Unregister unregisterOldNode = new Unregister(RegisterNodeWithOldVersion.NODE_ID, + EnumSet.allOf(NodeState.class), + ClusterMetadataService.instance().placementProvider()); + ClusterMetadataService.instance().commit(unregisterOldNode); + + assertEquals(ClusterMetadata.current().directory.commonSerializationVersion, CURRENT_METADATA_VERSION); + bytes = t.kind().toVersionedBytes(t); + try (DataInputBuffer buf = new DataInputBuffer(bytes, true)) { - throw new RuntimeException(e); + assertEquals(CURRENT_METADATA_VERSION, Version.fromInt(buf.readUnsignedVInt32())); } } - catch (UnknownHostException e) + catch (IOException e) { throw new RuntimeException(e); } @@ -172,28 +166,70 @@ public class RegisterTest extends TestBaseImpl { cluster.get(1).startup(); cluster.get(1).runOnInstance(() -> { + // Run a custom transformation to inject a fake node into the directory with a known id and an + // artificially lowered max supported serialization version + CustomTransformation.registerExtension(RegisterNodeWithOldVersion.NAME, RegisterNodeWithOldVersion.serializer); + CustomTransformation injectOldNode = new CustomTransformation(RegisterNodeWithOldVersion.NAME, + new RegisterNodeWithOldVersion()); + ClusterMetadataService.instance().commit(injectOldNode); + // Now trigger a snapshot which must be written to the snapshot using the old serialization version + Epoch epoch = ClusterMetadataService.instance().commit(TriggerSnapshot.instance).epoch; + // fetch the raw bytes of the snapshot we just serialized + ByteBuffer bytes = SystemKeyspace.getSnapshot(epoch); + assertNotNull(bytes); + // assert the prepended version matches + Version writtenVersion = null; try { - // Unregister to make directory empty - ClusterMetadataService.instance().commit(new Unregister(ClusterMetadata.current().myNodeId(), - EnumSet.allOf(NodeState.class), - ClusterMetadataService.instance().placementProvider())); - - // Register a ghost node with V0 (bypasses version check because directory is now empty). - // 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.100")), - TEST_LOCATION, - new NodeVersion(NodeVersion.CURRENT.cassandraVersion, Version.V0))); + writtenVersion = Version.fromInt(VIntCoding.readUnsignedVInt32(new DataInputBuffer(bytes, false))); } - catch (UnknownHostException e) + catch (IOException e) { throw new RuntimeException(e); } - ClusterMetadataService.instance().commit(TriggerSnapshot.instance); - + assertEquals(RegisterNodeWithOldVersion.METADATA_VERSION, writtenVersion); + // load the snapshot using the standard mechanism and assert it matches current cluster metadata ClusterMetadata cm = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots().getSnapshot(ClusterMetadata.current().epoch); cm.equals(ClusterMetadata.current()); }); } } + + // Custom transforms to lock/unlock an arbitrary set of ranges to + // avoid having to actually initiate some range movement + public static class RegisterNodeWithOldVersion implements Transformation, Serializable + { + public static final AsymmetricMetadataSerializer serializer = new AsymmetricMetadataSerializer() + { + @Override + public void serialize(Transformation t, DataOutputPlus out, Version version){} + @Override + public RegisterNodeWithOldVersion deserialize(DataInputPlus in, Version version) {return new RegisterNodeWithOldVersion();} + @Override + public long serializedSize(Transformation t, Version version) {return 0;} + }; + + public static final String NAME = "TestRegisterNodeWithOldVersion"; + public static final NodeId NODE_ID = new NodeId(99); + public static final Version METADATA_VERSION = Version.V0; + + @Override + public Kind kind() + { + return Kind.CUSTOM; + } + + @Override + public Result execute(ClusterMetadata metadata) + { + ClusterMetadata.Transformer transformer = metadata.transformer() + .unsafeRegisterForTesting(NODE_ID, + new NodeAddresses(addr(99)), + TEST_LOCATION, + new NodeVersion(NodeVersion.CURRENT.cassandraVersion, + METADATA_VERSION)); + return Transformation.success(transformer, LockedRanges.AffectedRanges.EMPTY); + } + } + }