This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 8f8c4ec09eed97e8c7e5afcc58c048bc32c6cc47 Merge: 3ace21c90d 896d1d6415 Author: Sam Tunnicliffe <[email protected]> AuthorDate: Mon Jun 29 18:17:54 2026 +0100 Merge branch 'cassandra-6.0' into trunk CHANGES.txt | 1 + .../apache/cassandra/db/SystemPeersValidator.java | 3 + .../schema/DistributedMetadataLogKeyspace.java | 31 ++-- .../apache/cassandra/schema/DistributedSchema.java | 63 ++++---- .../apache/cassandra/service/StorageService.java | 14 +- .../cassandra/tcm/AbstractLocalProcessor.java | 2 +- .../org/apache/cassandra/tcm/CMSOperations.java | 4 +- .../org/apache/cassandra/tcm/ClusterMetadata.java | 23 ++- .../cassandra/tcm/ClusterMetadataService.java | 64 +++++++- src/java/org/apache/cassandra/tcm/Startup.java | 33 +++-- .../cassandra/tcm/compatibility/GossipHelper.java | 4 +- .../tcm/listeners/LegacyStateListener.java | 7 +- .../tcm/listeners/UpgradeMigrationListener.java | 17 ++- .../org/apache/cassandra/tcm/log/LocalLog.java | 32 ++-- .../apache/cassandra/tcm/membership/Directory.java | 26 ++++ .../tcm/migration/CMSInitializationException.java | 27 ++++ .../tcm/migration/CMSInitializationRequest.java | 10 ++ .../apache/cassandra/tcm/migration/Election.java | 20 +++ .../cassandra/tcm/ownership/DataPlacements.java | 6 + .../tcm/transformations/cms/Initialize.java | 4 +- .../tcm/transformations/cms/PreInitialize.java | 41 +++--- .../ClusterMetadataSingleNodeUpgradeTest.java | 12 +- ...> ClusterMetadataUpgradeDC2InitializeTest.java} | 39 +++-- .../upgrade/ClusterMetadataUpgradeHarryTest.java | 2 + ...sterMetadataUpgradeInconsistentPeersV2Test.java | 114 ++++++++++++++ .../ClusterMetadataUpgradePeersHostIdsTest.java | 109 ++++++++++++++ .../upgrade/ClusterMetadataUpgradeTest.java | 10 +- ...lusterMetadataUpgradeUnexpectedFailureTest.java | 163 +++++++++++++++++---- .../unit/org/apache/cassandra/ServerTestUtils.java | 8 +- .../cassandra/schema/DistributedSchemaTest.java | 5 +- .../cassandra/tcm/log/DistributedLogStateTest.java | 7 +- .../tcm/log/LogListenerNotificationTest.java | 2 +- .../cassandra/utils/CassandraGenerators.java | 2 +- 33 files changed, 727 insertions(+), 178 deletions(-) diff --cc CHANGES.txt index b1f89bc51c,990378b6f3..6566e81353 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,9 -1,5 +1,10 @@@ -6.0-alpha2 +7.0 + * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975) + * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394) + * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) + * Add a guardrail for misprepared statements (CASSANDRA-21139) +Merged from 6.0: + * Defer creation of the system_cluster_metadata keyspace until CMS initialization (CASSANDRA-21477) * Support direct I/O for background SSTable writes (CASSANDRA-21134) * Relax assertion on partitioner instances in SinglePartitionReadCommand (CASSANDRA-21251) * Report cancelled read command execution to coordinator as a RequestFailure.TIMEOUT (CASSANDRA-21468) diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDC2InitializeTest.java index 9d108f41d0,48435ef789..9426c2332d --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDC2InitializeTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDC2InitializeTest.java @@@ -23,33 -27,38 +27,38 @@@ import org.slf4j.LoggerFactory import org.apache.cassandra.distributed.Constants; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInvokableInstance; - import org.apache.cassandra.service.StorageService; + import org.apache.cassandra.distributed.shared.ClusterUtils; + import org.apache.cassandra.distributed.shared.NetworkTopology; - import static org.junit.Assert.assertTrue; + import static org.junit.Assert.assertFalse; - /** - * @see org.apache.cassandra.tools.nodetool.CMSAdmin.InitializeCMS - */ - public class ClusterMetadataSingleNodeUpgradeTest extends UpgradeTestBase + public class ClusterMetadataUpgradeDC2InitializeTest extends UpgradeTestBase { + public static final Logger logger = LoggerFactory.getLogger(ClusterMetadataUpgradeDC2InitializeTest.class); @Test - public void testSingleNodeUpgrade() throws Throwable + public void testCMSInitializeOnDC2NodeAfterUpgrade() throws Throwable { new TestCase() - .nodes(1) - .nodesToUpgrade(1) + .nodes(2) + .nodesToUpgrade(1, 2) + .withNodeIdTopology(ImmutableMap.of(1, NetworkTopology.dcAndRack("dc1", "rack1"), + 2, NetworkTopology.dcAndRack("dc2", "rack2"))) .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .upgradesToCurrentFrom(v41) + .singleUpgradeToCurrentFrom(v50) .setup((cluster) -> { - cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':1}")); - cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + }) + .runBeforeClusterUpgrade(cluster -> { + cluster.forEach(node -> { + node.flush("system"); + }); }) .runAfterClusterUpgrade((cluster) -> { - assertTrue(((IInvokableInstance)cluster.get(1)).callOnInstance(() -> StorageService.instance.getRangeToAddressMap("system_cluster_metadata").isEmpty())); - cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); - // make sure we can execute transformations: - cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl with comment = 'hello123'")); + // Run cms initialize on the node in dc2 (node 2) instead of dc1. + cluster.get(2).nodetoolResult("cms", "initialize").asserts().success(); + + cluster.forEach(i -> assertFalse("node " + i.config().num() + " is still in MIGRATING STATE", + ClusterUtils.isMigrating((IInvokableInstance) i))); }).run(); } } diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeInconsistentPeersV2Test.java index 0000000000,8724403ed7..f4c0e10b1e mode 000000,100644..100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeInconsistentPeersV2Test.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeInconsistentPeersV2Test.java @@@ -1,0 -1,114 +1,114 @@@ + /* + * 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.upgrade; + + import java.util.Map; + + import org.junit.Test; + + import org.apache.cassandra.distributed.Constants; + import org.apache.cassandra.distributed.UpgradeableCluster; + import org.apache.cassandra.distributed.api.Feature; + import org.apache.cassandra.distributed.api.IInvokableInstance; + import org.apache.cassandra.distributed.api.IUpgradeableInstance; + import org.apache.cassandra.distributed.api.SimpleQueryResult; + import org.apache.cassandra.distributed.shared.NetworkTopology; + import org.apache.cassandra.locator.MetaStrategy; + import org.apache.cassandra.locator.NetworkTopologyStrategy; + import org.apache.cassandra.schema.ReplicationParams; + import org.apache.cassandra.tcm.ClusterMetadata; + + import static java.lang.String.format; + import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME; + import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertFalse; + import static org.junit.Assert.assertTrue; + + public class ClusterMetadataUpgradeInconsistentPeersV2Test extends UpgradeTestBase + { + @Test + public void upgradeWithInconsistentSystemPeersV2Test() throws Throwable + { + new TestCase() + .nodes(3) + .withNodeIdTopology(NetworkTopology.networkTopology(3, (i) -> NetworkTopology.dcAndRack("datacenter0" + i % 2, "rack0" + i))) + .nodesToUpgrade(1, 2, 3) + .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) + .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .upgradesToCurrentFrom(v41) ++ .singleUpgradeToCurrentFrom(v50) + .setup((cluster) -> { + // insert mismatching entries into system.peers_v2 table + for (int i = 1; i <= 3; i++) + { + cluster.get(i).executeInternal(format("insert into system.peers_v2 (peer, peer_port, data_center) " + + "values ('10.10.10.10', 7000, 'a%s')", i)); + cluster.get(i).flush("system"); + } + }) + .runAfterClusterUpgrade((cluster) -> { + // The system_cluster_metadata keyspace shouldn't be created until the CMS is intialized + assertNoMetaKeyspace(cluster); + + cluster.get(3).nodetoolResult("cms", "initialize").asserts().success(); + + // post-initialization, the replication params for system_cluster_metadata should be RF 1 in the DC of the + // first CMS member - node3 / datacenter01 + Map<String, String> actualReplication = Map.of("class", MetaStrategy.class.getName(), "datacenter01", "1"); + Map<String, String> fromSystemTable = Map.of("class", NetworkTopologyStrategy.class.getName(), "datacenter01", "1"); + assertReplicationParams(cluster, actualReplication, fromSystemTable); + }).run(); + } + + private static void assertNoMetaKeyspace(UpgradeableCluster cluster) + { + for (IUpgradeableInstance inst : cluster) + { + IInvokableInstance i = (IInvokableInstance) inst; + boolean found = i.callOnInstance(() -> ClusterMetadata.current().schema.getKeyspaces().containsKeyspace(METADATA_KEYSPACE_NAME)); + assertFalse("Metadata keyspace present on node" + i + " when it should not be", found); + + SimpleQueryResult res = inst.executeInternalWithResult("select replication from system_schema.keyspaces " + + "where keyspace_name = ?", + METADATA_KEYSPACE_NAME); + assertFalse(res.hasNext()); + } + } + + private static void assertReplicationParams(UpgradeableCluster cluster, + Map<String, String> expectedActual, + Map<String, String> expectedInSystemTable) + { + for (IUpgradeableInstance inst : cluster) + { + IInvokableInstance i = (IInvokableInstance) inst; + Map<String, String> rs = i.callOnInstance(() -> { + ReplicationParams r = ClusterMetadata.current().schema.getKeyspaceMetadata("system_cluster_metadata").params.replication; + return r.asMap(); + }); + assertEquals(rs, expectedActual); + + SimpleQueryResult res = inst.executeInternalWithResult("select replication from system_schema.keyspaces " + + "where keyspace_name = ?", + METADATA_KEYSPACE_NAME); + assertTrue(res.hasNext()); + Map<String, String> replication = res.next().get("replication"); + assertEquals(replication, expectedInSystemTable); + } + } + } diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePeersHostIdsTest.java index 0000000000,e7f7a7a054..2ef0e4631c mode 000000,100644..100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePeersHostIdsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePeersHostIdsTest.java @@@ -1,0 -1,109 +1,109 @@@ + /* + * 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.upgrade; + + import java.util.HashMap; + import java.util.Map; + import java.util.UUID; + + import org.junit.Test; + + import org.apache.cassandra.distributed.Constants; + import org.apache.cassandra.distributed.api.Feature; + import org.apache.cassandra.distributed.api.IUpgradeableInstance; + import org.apache.cassandra.distributed.api.Row; + import org.apache.cassandra.distributed.api.SimpleQueryResult; + import org.apache.cassandra.distributed.impl.TestEndpointCache; + import org.apache.cassandra.locator.InetAddressAndPort; + import org.apache.cassandra.tcm.membership.NodeId; + + import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertFalse; + + public class ClusterMetadataUpgradePeersHostIdsTest extends UpgradeTestBase + { + @Test + public void upgradeHostIdUpdateTest() throws Throwable + { + Map<InetAddressAndPort, UUID> preUpgradeIDs = new HashMap<>(); + new TestCase() + .nodes(3) + .nodesToUpgrade(1, 2, 3) + .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) + .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .upgradesToCurrentFrom(v41) ++ .singleUpgradeToCurrentFrom(v50) + .setup((cluster) -> { + // combine the system.peers_v2 entries from each instance pre-upgrade + for (int i = 1; i <= 3; i++) + preUpgradeIDs.putAll(getHostIdsFromSystemPeersV2(cluster.get(i))); + + assertEquals(3, preUpgradeIDs.size()); + + for (UUID hostId : preUpgradeIDs.values()) + assertFalse(NodeId.isValidNodeId(hostId)); + }) + .runAfterClusterUpgrade((cluster) -> { + for (int i = 1; i <= 3; i++) + { + // system.peers/peers_v2 should still contain the pre-upgrade ids + IUpgradeableInstance inst = cluster.get(i); + Map<InetAddressAndPort, UUID> expected = new HashMap<>(preUpgradeIDs); + expected.remove(TestEndpointCache.toCassandraInetAddressAndPort(inst.config().broadcastAddress())); + assertEquals(expected, getHostIdsFromSystemPeersV2(cluster.get(i))); + } + + // initialize the CMS and fetch new ids from the ClusterMetadata Directory + cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); + Map<InetAddressAndPort, UUID> postUpgradeIDs = getHostIdsFromClusterMetadata(cluster.get(1)); + + for (int i = 1; i <= 3; i++) + { + // assert system.peers/peers_v2 now contain the post-upgrade ids + IUpgradeableInstance inst = cluster.get(i); + Map<InetAddressAndPort, UUID> expected = new HashMap<>(postUpgradeIDs); + expected.remove(TestEndpointCache.toCassandraInetAddressAndPort(inst.config().broadcastAddress())); + assertEquals(expected, getHostIdsFromSystemPeersV2(cluster.get(i))); + } + }).run(); + } + + private static Map<InetAddressAndPort, UUID> getHostIdsFromSystemPeersV2(IUpgradeableInstance instance) + { + Map<InetAddressAndPort, UUID> hostIds = new HashMap<>(); + SimpleQueryResult res = instance.executeInternalWithResult("select peer, peer_port, host_id from system.peers_v2"); + while(res.hasNext()) + { + Row row = res.next(); + hostIds.put(InetAddressAndPort.getByAddressOverrideDefaults(row.get(0), row.getInteger(1)), row.getUUID(2)); + } + return hostIds; + } + + private static Map<InetAddressAndPort, UUID> getHostIdsFromClusterMetadata(IUpgradeableInstance instance) + { + Map<InetAddressAndPort, UUID> hostIds = new HashMap<>(); + SimpleQueryResult res = instance.executeInternalWithResult("select broadcast_address, broadcast_port, host_id from system_views.cluster_metadata_directory"); + while(res.hasNext()) + { + Row row = res.next(); + hostIds.put(InetAddressAndPort.getByAddressOverrideDefaults(row.get(0), row.getInteger(1)), row.getUUID(2)); + } + return hostIds; + } + } diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeUnexpectedFailureTest.java index 746f561523,d7bb080ca9..052bbd9300 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeUnexpectedFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeUnexpectedFailureTest.java @@@ -45,24 -74,79 +74,79 @@@ public class ClusterMetadataUpgradeUnex { Consumer<UpgradeableCluster.Builder > builderUpdater = builder -> builder.withInstanceInitializer(BBInstaller::installUpgradeVersionBB); new TestCase() - .nodes(3) - .nodesToUpgrade(1, 2, 3) - .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) - .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .singleUpgradeToCurrentFrom(v41) - .withBuilder(builderUpdater) - .setup((cluster) -> { - cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}")); - cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); - }) - .runAfterClusterUpgrade((cluster) -> { - // we injected a BB helper to trigger an unexpected failure on the first attempt at initialization. - // i.e. an exception that must be caught as opposed to a mismatch in metadata or down peer. - cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure().errorContains("Something unexpected went wrong"); - // handling the failure should have included cleaning up any state so that another attempt can be - // made, which this time should succeed. - cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); - }).run(); + .nodes(3) + .nodesToUpgrade(1, 2, 3) + .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP) + .set(Constants.KEY_DTEST_FULL_STARTUP, true)) - .upgradesToCurrentFrom(v41) ++ .singleUpgradeToCurrentFrom(v50) + .withBuilder(builderUpdater) + .setup((cluster) -> { + BBState.failBeforePreInitialize.set(failBeforePreInit); + BBState.failAfterPreInitialize.set(failAfterPreInit); + cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}")); + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + for (int i = 0; i < 10; i++) + { + cluster.get(1).coordinator().execute("INSERT into " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)", + ConsistencyLevel.ALL, + i, i, i); + } + }) + .runAfterClusterUpgrade((cluster) -> { + String oldHostId = getHostId(cluster.get(1)); + // we injected a BB helper to trigger an unexpected failure on the first attempt at initialization. + // i.e. an exception that must be caught as opposed to a mismatch in metadata or down peer. + cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure().errorContains(expectedError); + assertEquals(oldHostId, getHostId(cluster.get(1))); + + // handling the failure should have included cleaning up any state so that another attempt can be + // made, which this time should succeed. + for(IUpgradeableInstance inst : cluster) + inst.nodetoolResult("cms").asserts().success().stdoutContains("Service State: GOSSIP"); + // Basic smoke test + for (int i = 0; i < 10; i++) + { + Object [][] rows = cluster.get(1) + .coordinator() + .execute("SELECT v from " + KEYSPACE + ".tbl WHERE pk = ?", + ConsistencyLevel.ALL, i); + assertEquals(1, rows.length); + assertEquals(i, rows[0][0]); + } + + // Make sure that no trace of the metadata keyspace is present after the CMS initialization failure + assertSchemaTablesContent(cluster, true); + + Object[][] rows = cluster.get(1).coordinator().execute("DESCRIBE FULL SCHEMA", ConsistencyLevel.NODE_LOCAL); + for (Object[] row : rows) + assertFalse(row[0].toString().equalsIgnoreCase(METADATA_KEYSPACE_NAME)); + + assertEquals(oldHostId, getHostId(cluster.get(1))); + // A subsequent initialization should succeed + cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); + assertNotEquals(oldHostId, getHostId(cluster.get(1))); + assertSchemaTablesContent(cluster, false); + }).run(); + } + + private static void assertSchemaTablesContent(UpgradeableCluster cluster, boolean expectEmpty) + { + for (String schemaTable : new String[] { KEYSPACES, TABLES, COLUMNS }) + { + Object[][] rows = cluster.get(1) + .coordinator() + .execute("SELECT * FROM " + SCHEMA_KEYSPACE_NAME + "." + schemaTable + " WHERE keyspace_name = ?" , + ConsistencyLevel.ALL, METADATA_KEYSPACE_NAME); + if (expectEmpty) + assertEquals(0, rows.length); + else + assertTrue(rows.length >= 1); + } + } + + private static String getHostId(IUpgradeableInstance i) + { + return ((IInvokableInstance)i).callOnInstance(() -> SystemKeyspace.getLocalHostId().toString()); } public static class BBInstaller --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
