mirror of https://github.com/apache/cassandra
[CASSANDRA-20736] Rewrite RegisterTest to use a custom transformation to setup Directory
This commit is contained in:
Sam Tunnicliffe
parent
9cee800b38
commit
097ea5a59f
|
|
@ -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<Transformation, RegisterNodeWithOldVersion> serializer = new AsymmetricMetadataSerializer<Transformation, RegisterNodeWithOldVersion>()
|
||||
{
|
||||
@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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue