natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Correctly update peers tables following replacement 

Patch by Sam Tunnicliffe; reviewed by David Capwell for CASSANDRA-19782
This commit is contained in:
Sam Tunnicliffe 2024-07-18 19:23:46 +01:00
parent 5434cede31
commit 32755cabfa
3 changed files with 38 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1 5.1
* Update legacy peers tables during node replacement (CASSANDRA-19782)
* Refactor ColumnCondition (CASSANDRA-19620) * Refactor ColumnCondition (CASSANDRA-19620)
* Allow configuring log format for Audit Logs (CASSANDRA-19792) * Allow configuring log format for Audit Logs (CASSANDRA-19792)
* Support for noboolean rpm (centos7 compatible) packages removed (CASSANDRA-19787) * Support for noboolean rpm (centos7 compatible) packages removed (CASSANDRA-19787)

1
src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java
View File

@ -158,6 +158,7 @@ public class LegacyStateListener implements ChangeListener.Async
logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement); logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement);
} }
Gossiper.instance.mergeNodeToGossip(change, next, tokens); Gossiper.instance.mergeNodeToGossip(change, next, tokens);
PeersTable.updateLegacyPeerTable(change, prev, next);
} }
} }
else else

36
test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
View File

@ -19,8 +19,11 @@
package org.apache.cassandra.distributed.test.hostreplacement; package org.apache.cassandra.distributed.test.hostreplacement;
import java.io.IOException; import java.io.IOException;
import java.net.InetAddress;
import java.util.Arrays; import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List; import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.junit.Test; import org.junit.Test;
@ -41,11 +44,15 @@ import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.assertj.core.api.Assertions; import org.assertj.core.api.Assertions;
import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK;
import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY; import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
import static org.apache.cassandra.distributed.shared.ClusterUtils.assertInRing; import static org.apache.cassandra.distributed.shared.ClusterUtils.assertInRing;
import static org.apache.cassandra.distributed.shared.ClusterUtils.assertRingIs; import static org.apache.cassandra.distributed.shared.ClusterUtils.assertRingIs;
import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingHealthy; import static org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingHealthy;
@ -115,6 +122,7 @@ public class HostReplacementTest extends TestBaseImpl
String replacingNodeAddress = replacingNode.config().broadcastAddress().getHostString(); String replacingNodeAddress = replacingNode.config().broadcastAddress().getHostString();
validateGossipStatusNormal(seed, replacingNodeAddress); validateGossipStatusNormal(seed, replacingNodeAddress);
validateGossipStatusNormal(replacingNode, replacingNodeAddress); validateGossipStatusNormal(replacingNode, replacingNodeAddress);
validatePeersTables(seed, replacingNode, even);
} }
} }
@ -266,4 +274,32 @@ public class HostReplacementTest extends TestBaseImpl
}); });
} }
static void validatePeersTables(IInvokableInstance seed, IInvokableInstance replacingNode, TokenSupplier tokens)
{
InetAddress replacementAddress = replacingNode.config().broadcastAddress().getAddress();
NodeId replacementId = new NodeId(3);
UUID schemaVersion = seed.callOnInstance(() -> ClusterMetadata.current().schema.getVersion());
String releaseVersion = seed.callOnInstance(() -> ClusterMetadata.current().directory.version(ClusterMetadata.current().myNodeId()).cassandraVersion.toString());
LinkedHashSet<String> replacementTokens = new LinkedHashSet<>(tokens.tokens(2));
String datacenter = "datacenter0";
String rack = "rack0";
assertRows(seed.executeInternal("SELECT * from system.peers"),
row(replacementAddress, datacenter, replacementId.toUUID(),
replacementAddress, rack, releaseVersion, replacementAddress,
schemaVersion, replacementTokens));
assertRows(seed.executeInternal("SELECT * from system.peers_v2"),
row(replacementAddress, 7012, datacenter, replacementId.toUUID(),
replacementAddress, 9042, replacementAddress, 7012, rack, releaseVersion,
schemaVersion, replacementTokens));
// system_views.peers contains both remote and local node info
InetAddress seedAddress = seed.config().broadcastAddress().getAddress();
assertRows(seed.executeInternal("SELECT * from system_views.peers"),
rows(row(seedAddress, 7012, datacenter, new NodeId(1).toUUID(),
seedAddress, 9042, seedAddress, 7012, rack, releaseVersion,
schemaVersion, NodeState.JOINED.toString(), new LinkedHashSet<>(tokens.tokens(1))),
row(replacementAddress, 7012, datacenter, replacementId.toUUID(),
replacementAddress, 9042, replacementAddress, 7012, rack, releaseVersion,
schemaVersion, NodeState.JOINED.toString(), replacementTokens)));
}
} }