diff --git a/src/java/org/apache/cassandra/index/IndexStatusManager.java b/src/java/org/apache/cassandra/index/IndexStatusManager.java index 1ffb53d004..e84ee98c7b 100644 --- a/src/java/org/apache/cassandra/index/IndexStatusManager.java +++ b/src/java/org/apache/cassandra/index/IndexStatusManager.java @@ -28,7 +28,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; -import java.util.UUID; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -59,6 +58,7 @@ import org.apache.cassandra.schema.SystemDistributedKeyspace; import org.apache.cassandra.serializers.MarshalException; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.ExecutorUtils; @@ -83,8 +83,6 @@ public class IndexStatusManager private static final int MAX_GOSSIP_VALUE_SIZE = 65535; - public static final String TABLE_FALLBACK_MARKER = "TABLE_FALLBACK"; - private volatile long lastPollTimestampMillis = 0; private ScheduledFuture pollFuture; @@ -96,7 +94,7 @@ public class IndexStatusManager /** * A map of per-endpoint index statuses: the key of inner map is the identifier "keyspace.index" */ - public final Map> peerIndexStatus = new HashMap<>(); + public final Map> peerIndexStatus = new HashMap<>(); private IndexStatusManager() {} @@ -182,18 +180,18 @@ public class IndexStatusManager if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort())) return; + NodeId nodeId = ClusterMetadata.current().directory.peerId(endpoint); + + if (nodeId == null) + { + logger.warn("Ignoring index status from unknown endpoint {}", endpoint); + return; + } + Map indexStatusMap; - if (versionedValue.value.equals(TABLE_FALLBACK_MARKER)) - { - indexStatusMap = SystemDistributedKeyspace.allIndexStatusesForHost(StorageService.instance.getHostIdForEndpoint(endpoint)); - } - else - { - indexStatusMap = statusMapFromString(versionedValue); - } - - Map oldStatus = peerIndexStatus.put(endpoint, indexStatusMap); + indexStatusMap = statusMapFromString(versionedValue); + Map oldStatus = peerIndexStatus.put(nodeId, indexStatusMap); Map updated = updatedIndexStatuses(oldStatus, indexStatusMap); Set removed = removedIndexStatuses(oldStatus, indexStatusMap); if (!updated.isEmpty() || !removed.isEmpty()) @@ -254,40 +252,36 @@ public class IndexStatusManager { try { - Map statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(), - k -> new HashMap<>()); + NodeId localNodeId = ClusterMetadata.current().myNodeId(); + if (localNodeId == null) + return; + Map statusMap = peerIndexStatus.computeIfAbsent(localNodeId, k -> new HashMap<>()); String keyspaceIndex = identifier(keyspace, index); - UUID localHostId = StorageService.instance.getLocalHostUUID(); CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion; + boolean shouldWriteToIndexTables = shouldWriteToIndexTables(minVersion); if (status == Index.Status.DROPPED) { statusMap.remove(keyspaceIndex); - if (localHostId != null) - statusPropagationExecutor.submit(() -> { - if (shouldWriteToIndexTables(minVersion)) - { - SystemDistributedKeyspace.setIndexRemoved(localHostId, keyspace, index); - SystemDistributedKeyspace.recordIndexEvent(localHostId, keyspace, index, status); - } - }); + if (shouldWriteToIndexTables) + { + SystemDistributedKeyspace.setIndexRemoved(localNodeId, keyspace, index); + SystemDistributedKeyspace.recordIndexEvent(localNodeId, keyspace, index, status); + } } else { statusMap.put(keyspaceIndex, status); - if (localHostId != null) - statusPropagationExecutor.submit(() -> { - if (shouldWriteToIndexTables(minVersion)) - { - SystemDistributedKeyspace.updateIndexStatus(localHostId, keyspace, index, status); - SystemDistributedKeyspace.recordIndexEvent(localHostId, keyspace, index, status); - } - }); + if (shouldWriteToIndexTables) + { + SystemDistributedKeyspace.updateIndexStatus(localNodeId, keyspace, index, status); + SystemDistributedKeyspace.recordIndexEvent(localNodeId, keyspace, index, status); + } } // Only propagate via gossip when the gossiper is enabled and the cluster is not fully on 6.0+. // Once all nodes are on 6.0+, index status is propagated via table polling instead. - if (Gossiper.instance.isEnabled() && !shouldWriteToIndexTables(minVersion)) + if (Gossiper.instance.isEnabled() && !shouldWriteToIndexTables) { // Versions 5.0.0 through 5.0.2 use a much more bloated format that duplicates keyspace names // and writes full status names instead of their numeric codes. If the minimum cluster version is @@ -296,26 +290,21 @@ public class IndexStatusManager : toSerializedFormat(statusMap); byte[] utf8Bytes = newSerializedStatusMap.getBytes(StandardCharsets.UTF_8); - String gossipPayload; - if (utf8Bytes.length > MAX_GOSSIP_VALUE_SIZE) { logger.error("Index status gossip payload size ({} bytes) exceeds limit ({} bytes), please consider removing unwanted indexes.", utf8Bytes.length, MAX_GOSSIP_VALUE_SIZE); - gossipPayload = TABLE_FALLBACK_MARKER; - } - else - { - if (utf8Bytes.length > MAX_GOSSIP_VALUE_SIZE * 0.8) - logger.warn("Index status gossip payload size ({} bytes) approaching the limit ({} bytes), please consider removing unwanted indexes.", - utf8Bytes.length, MAX_GOSSIP_VALUE_SIZE); - gossipPayload = newSerializedStatusMap; + return; } + if (utf8Bytes.length > MAX_GOSSIP_VALUE_SIZE * 0.8) + logger.warn("Index status gossip payload size ({} bytes) approaching the limit ({} bytes), please consider removing unwanted indexes.", + utf8Bytes.length, MAX_GOSSIP_VALUE_SIZE); + statusPropagationExecutor.submit(() -> { // schedule gossiper update asynchronously to avoid potential deadlock when another thread is holding // gossiper taskLock. - VersionedValue value = StorageService.instance.valueFactory.indexStatus(gossipPayload); + VersionedValue value = StorageService.instance.valueFactory.indexStatus(newSerializedStatusMap); Gossiper.instance.addLocalApplicationState(ApplicationState.INDEX_STATUS, value); }); } @@ -342,18 +331,6 @@ public class IndexStatusManager return shouldWriteToIndexTables(minVersion); } - @VisibleForTesting - void processEventsForTesting(UntypedResultSet results) - { - processEvents(results); - } - - @VisibleForTesting - void resetLastPollTimestamp() - { - lastPollTimestampMillis = 0; - } - /** * Serializes as a JSON string the status of the indexes in the provided map. *

@@ -390,7 +367,10 @@ public class IndexStatusManager @VisibleForTesting public synchronized Index.Status getIndexStatus(InetAddressAndPort peer, String keyspace, String index) { - return peerIndexStatus.getOrDefault(peer, Collections.emptyMap()) + NodeId nodeId = ClusterMetadata.current().directory.peerId(peer); + if (nodeId == null) + return Index.Status.UNKNOWN; + return peerIndexStatus.getOrDefault(nodeId, Collections.emptyMap()) .getOrDefault(identifier(keyspace, index), Index.Status.UNKNOWN); } @@ -435,14 +415,10 @@ public class IndexStatusManager { try { - Map> allStatuses = SystemDistributedKeyspace.allIndexStatuses(); - for (Map.Entry> entry : allStatuses.entrySet()) + Map> allStatuses = SystemDistributedKeyspace.allIndexStatuses(); + for (Map.Entry> entry : allStatuses.entrySet()) { - InetAddressAndPort endpoint = StorageService.instance.getEndpointForHostId(entry.getKey()); - if (endpoint == null) - continue; - - peerIndexStatus.putIfAbsent(endpoint, entry.getValue()); + peerIndexStatus.putIfAbsent(entry.getKey(), entry.getValue()); } logger.info("Loaded index statuses from system table for {} peers", allStatuses.size()); } @@ -459,14 +435,10 @@ public class IndexStatusManager { try { - Map> allStatuses = SystemDistributedKeyspace.allIndexStatuses(); - for (Map.Entry> entry : allStatuses.entrySet()) + Map> allStatuses = SystemDistributedKeyspace.allIndexStatuses(); + for (Map.Entry> entry : allStatuses.entrySet()) { - InetAddressAndPort endpoint = StorageService.instance.getEndpointForHostId(entry.getKey()); - if (endpoint == null) - continue; - - peerIndexStatus.put(endpoint, entry.getValue()); + peerIndexStatus.put(entry.getKey(), entry.getValue()); } logger.info("Refreshed index statuses from system table for {} peers", allStatuses.size()); } @@ -499,15 +471,19 @@ public class IndexStatusManager private synchronized void pollIndexEvents() { - if (lastPollTimestampMillis == 0) + try { - refreshFromFullTable(); - lastPollTimestampMillis = Clock.Global.currentTimeMillis(); - } - else - { - try + if (lastPollTimestampMillis == 0) { + refreshFromFullTable(); + String today = LocalDate.now(ZoneOffset.UTC).toString(); + UntypedResultSet todayResults = SystemDistributedKeyspace.queryIndexEvents(today, 0); + long newestEventTime = todayResults != null ? processEvents(todayResults) : 0; + lastPollTimestampMillis = newestEventTime != 0 ? newestEventTime : Clock.Global.currentTimeMillis(); + } + else + { + long newestPollTimestampMillis = 0; String today = LocalDate.now(ZoneOffset.UTC).toString(); String lastPollDate = Instant.ofEpochMilli(lastPollTimestampMillis).atZone(ZoneOffset.UTC).toLocalDate().toString(); @@ -516,32 +492,32 @@ public class IndexStatusManager // midnight crossed so get remaining events from last day. UntypedResultSet yesterdayResults = SystemDistributedKeyspace.queryIndexEvents(lastPollDate, lastPollTimestampMillis); if (yesterdayResults != null) - processEvents(yesterdayResults); + newestPollTimestampMillis = processEvents(yesterdayResults); } UntypedResultSet todayResults = SystemDistributedKeyspace.queryIndexEvents(today, lastPollTimestampMillis); if (todayResults != null) - processEvents(todayResults); + newestPollTimestampMillis = processEvents(todayResults); - lastPollTimestampMillis = Clock.Global.currentTimeMillis(); - } - catch (Exception e) - { - logger.warn("Unable to load index events from system table: {}", e.getMessage()); + if (newestPollTimestampMillis != 0) + lastPollTimestampMillis = newestPollTimestampMillis; } } + catch (Exception e) + { + logger.warn("Unable to load index events from system table: {}", e.getMessage()); + } } - private void processEvents(UntypedResultSet results) + private long processEvents(UntypedResultSet results) { + NodeId localNodeId = ClusterMetadata.current().myNodeId(); + long newestEventTime = 0; for (UntypedResultSet.Row row : results) { - UUID hostId = row.getUUID("host_id"); - if ((hostId == null) || hostId.equals(StorageService.instance.getLocalHostUUID())) - continue; - - InetAddressAndPort endpoint = StorageService.instance.getEndpointForHostId(hostId); - if (endpoint == null) + newestEventTime = row.getTimestamp("event_time").getTime(); + NodeId nodeId = new NodeId(row.getInt("node_id")); + if (nodeId.equals(localNodeId)) continue; String indexName = row.getString("index_name"); @@ -549,14 +525,15 @@ public class IndexStatusManager if (status == Index.Status.DROPPED) { - Map statusMap = peerIndexStatus.get(endpoint); + Map statusMap = peerIndexStatus.get(nodeId); if (statusMap != null) statusMap.remove(indexName); } else { - peerIndexStatus.computeIfAbsent(endpoint, k -> new HashMap<>()).put(indexName, status); + peerIndexStatus.computeIfAbsent(nodeId, k -> new HashMap<>()).put(indexName, status); } } + return newestEventTime; } } diff --git a/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java b/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java index 3f842687b8..cc0d7ba95d 100644 --- a/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java @@ -61,6 +61,7 @@ import org.apache.cassandra.index.Index; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.CommonRange; import org.apache.cassandra.repair.messages.RepairOption; +import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.TimeUUID; @@ -220,11 +221,11 @@ public final class SystemDistributedKeyspace parse(COMPRESSION_DICTIONARIES, "Compression dictionaries for applicable tables", COMPRESSION_DICTIONARIES_CQL).build(); public static final String INDEX_BUILD_STATUS_CQL = "CREATE TABLE IF NOT EXISTS %s (" + - "host_id uuid," + + "node_id int," + "keyspace_name text," + "index_name text," + "status text," + - "PRIMARY KEY (host_id, keyspace_name, index_name))"; + "PRIMARY KEY (node_id, keyspace_name, index_name))"; private static final TableMetadata IndexBuildStatus = parse(INDEX_BUILD_STATUS, "Index build status", INDEX_BUILD_STATUS_CQL).build(); @@ -232,10 +233,10 @@ public final class SystemDistributedKeyspace "date text," + "event_time timestamp," + "index_name text," + - "host_id UUID," + + "node_id int," + "event text," + - "PRIMARY KEY (date, event_time, index_name, host_id)) " + - "WITH CLUSTERING ORDER BY (event_time DESC)"; + "PRIMARY KEY (date, event_time, index_name, node_id)) " + + "WITH CLUSTERING ORDER BY (event_time ASC)"; private static final TableMetadata IndexEventsTable = parse(INDEX_EVENTS, "Index events for applicable tables", INDEX_EVENTS_CQL) @@ -443,34 +444,55 @@ public final class SystemDistributedKeyspace forceBlockingFlush(VIEW_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED); } - public static void updateIndexStatus(UUID hostId, String keyspace, String index, Index.Status status) + public static void updateIndexStatus(NodeId nodeId, String keyspace, String index, Index.Status status) { - String query = "INSERT INTO %s.%s (host_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)"; - QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), - ConsistencyLevel.ONE, - Lists.newArrayList(bytes(hostId), - bytes(keyspace), - bytes(index), - bytes(status.toString()))); + String query = "INSERT INTO %s.%s (node_id, keyspace_name, index_name, status) VALUES (?, ?, ?, ?)"; + try + { + QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), + ConsistencyLevel.QUORUM, nodeId.id(), keyspace, index, status.toString()); + } + catch (Exception e) + { + logger.warn("Failed to update index status with QUORUM for {}.{}, retrying with ONE", keyspace, index); + QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), + ConsistencyLevel.ONE, nodeId.id(), keyspace, index, status.toString()); + } } - public static void setIndexRemoved(UUID hostId, String keyspaceName, String indexName) + public static void setIndexRemoved(NodeId nodeId, String keyspaceName, String indexName) { - String buildReq = "DELETE FROM %s.%s WHERE host_id = ? AND keyspace_name = ? AND index_name = ?"; - QueryProcessor.executeInternal(format(buildReq, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), - hostId, keyspaceName, indexName); + String buildReq = format("DELETE FROM %s.%s WHERE node_id = ? AND keyspace_name = ? AND index_name = ?", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS); + try + { + QueryProcessor.execute(buildReq, ConsistencyLevel.QUORUM, nodeId.id(), keyspaceName, indexName); + } + catch (Exception e) + { + logger.warn("Failed to remove index status with QUORUM for {}.{}, retrying with ONE", keyspaceName, indexName); + QueryProcessor.execute(buildReq, ConsistencyLevel.ONE, nodeId.id(), keyspaceName, indexName); + } forceBlockingFlush(INDEX_BUILD_STATUS, ColumnFamilyStore.FlushReason.INTERNALLY_FORCED); } - public static Map> allIndexStatuses() + public static Map> allIndexStatuses() { - String query = "SELECT host_id, keyspace_name, index_name, status FROM %s.%s"; + String query = format("SELECT node_id, keyspace_name, index_name, status FROM %s.%s", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS); UntypedResultSet results; try { - results = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), - ConsistencyLevel.ONE); + try + { + results = QueryProcessor.execute(query, ConsistencyLevel.QUORUM); + } + catch (Exception e) + { + logger.warn("Failed to load index statuses with QUORUM, retrying with ONE"); + results = QueryProcessor.execute(query, ConsistencyLevel.ONE); + } } catch (Exception e) { @@ -478,32 +500,39 @@ public final class SystemDistributedKeyspace return Collections.emptyMap(); } - Map> allStatuses = new HashMap<>(); + Map> allStatuses = new HashMap<>(); for (UntypedResultSet.Row row : results) { - UUID hostId = row.getUUID("host_id"); + NodeId nodeId = new NodeId(row.getInt("node_id")); String identifier = row.getString("keyspace_name") + '.' + row.getString("index_name"); Index.Status status = Index.Status.valueOf(row.getString("status")); - allStatuses.computeIfAbsent(hostId, k -> new HashMap<>()).put(identifier, status); + allStatuses.computeIfAbsent(nodeId, k -> new HashMap<>()).put(identifier, status); } return allStatuses; } - public static Map allIndexStatusesForHost(UUID hostId) + public static Map allIndexStatusesForHost(NodeId nodeId) { - String query = "SELECT keyspace_name, index_name, status FROM %s.%s WHERE host_id = ?"; + String query = format("SELECT keyspace_name, index_name, status FROM %s.%s WHERE node_id = ?", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS); UntypedResultSet results; try { - results = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_BUILD_STATUS), - ConsistencyLevel.ONE, - hostId); + try + { + results = QueryProcessor.execute(query, ConsistencyLevel.QUORUM, nodeId.id()); + } + catch (Exception e) + { + logger.warn("Failed to load index statuses with QUORUM for host {}, retrying with ONE", nodeId); + results = QueryProcessor.execute(query, ConsistencyLevel.ONE, nodeId.id()); + } } catch (Exception e) { - logger.warn("Unable to load index statuses from system table for host {}: {}", hostId, e.getMessage()); + logger.warn("Unable to load index statuses from system table for host {}: {}", nodeId, e.getMessage()); return Collections.emptyMap(); } @@ -517,34 +546,33 @@ public final class SystemDistributedKeyspace return statuses; } - public static void recordIndexEvent(UUID hostId, String keyspace, String index, Index.Status status) + public static void recordIndexEvent(NodeId nodeId, String keyspace, String index, Index.Status status) { - String query = "INSERT INTO %s.%s (date, event_time, index_name, host_id, event) VALUES (?, to_timestamp(now()), ?, ?, ?)"; - QueryProcessor.process(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_EVENTS), - ConsistencyLevel.ONE, - Lists.newArrayList(bytes(LocalDate.now(ZoneOffset.UTC).toString()), - bytes(keyspace + '.' + index), - bytes(hostId), - bytes(status.toString()))); + String query = format("INSERT INTO %s.%s (date, event_time, index_name, node_id, event) VALUES (?, to_timestamp(now()), ?, ?, ?)", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_EVENTS); + QueryProcessor.execute(query, ConsistencyLevel.QUORUM, + LocalDate.now(ZoneOffset.UTC).toString(), keyspace + '.' + index, nodeId.id(), status.toString()); } public static UntypedResultSet queryIndexEvents(String date, long sinceTimestampMillis) { - String query = "SELECT index_name, host_id, event FROM %s.%s WHERE date = ? AND event_time > ? "; - UntypedResultSet status; + String query = format("SELECT index_name, node_id, event, event_time FROM %s.%s WHERE date = ? AND event_time > ?", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_EVENTS); try { - status = QueryProcessor.execute(format(query, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, INDEX_EVENTS), - ConsistencyLevel.ONE, - date, - new java.util.Date(sinceTimestampMillis)); + return QueryProcessor.execute(query, ConsistencyLevel.QUORUM, date, new java.util.Date(sinceTimestampMillis)); } catch (Exception e) { - return null; + try + { + return QueryProcessor.execute(query, ConsistencyLevel.ONE, date, new java.util.Date(sinceTimestampMillis)); + } + catch (Exception ex) + { + return null; + } } - - return status; } /** diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java index bde10886b1..3cc7cc7017 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java @@ -48,6 +48,8 @@ import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SystemDistributedKeyspace; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.utils.FBUtilities; import static net.bytebuddy.matcher.ElementMatchers.named; @@ -75,7 +77,8 @@ public class IndexAvailabilityTest extends TestBaseImpl public void verifyIndexStatusPropagation() throws Exception { try (Cluster cluster = init(Cluster.build(2) - .withConfig(config -> config.with(GOSSIP).with(NETWORK)) + .withConfig(config -> config.with(GOSSIP).with(NETWORK) + .set("index_status_poll_interval_in_seconds", "1")) .start())) { verifyIndexStatusPropagation(cluster); @@ -202,7 +205,8 @@ public class IndexAvailabilityTest extends TestBaseImpl { try (Cluster cluster = init(Cluster.build(3) .withConfig(config -> config.with(GOSSIP) - .with(NETWORK)) + .with(NETWORK) + .set("index_status_poll_interval_in_seconds", "1")) .start())) { String ks2 = "ks2"; @@ -447,7 +451,7 @@ public class IndexAvailabilityTest extends TestBaseImpl cluster.get(1).runOnInstance(() -> { Map localStatusMap = IndexStatusManager.instance.peerIndexStatus - .computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(), k -> new HashMap<>()); + .computeIfAbsent(ClusterMetadata.current().myNodeId(), k -> new HashMap<>()); for (int ks = 0; ks < 100; ks++) for (int idx = 0; idx < 200; idx++) @@ -481,8 +485,7 @@ public class IndexAvailabilityTest extends TestBaseImpl waitForIndexingStatus(cluster.get(2), ks, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED); cluster.get(1).runOnInstance(() -> { - java.util.Map> allStatuses = - SystemDistributedKeyspace.allIndexStatuses(); + Map> allStatuses = SystemDistributedKeyspace.allIndexStatuses(); assertTrue("index_build_status table should be empty in mixed-version cluster, but has " + allStatuses.size() + " entries", allStatuses.isEmpty()); }); @@ -491,8 +494,8 @@ public class IndexAvailabilityTest extends TestBaseImpl waitForIndexingStatus(cluster.get(2), ks, index1, cluster.get(1), Index.Status.BUILD_FAILED); cluster.get(1).runOnInstance(() -> { - java.util.Map> allStatuses = - SystemDistributedKeyspace.allIndexStatuses(); + Map> allStatuses = + SystemDistributedKeyspace.allIndexStatuses(); assertTrue("index_build_status table should still be empty in mixed-version cluster", allStatuses.isEmpty()); }); diff --git a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java index 8d8c776c2d..ad714f06de 100644 --- a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java +++ b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java @@ -28,13 +28,16 @@ import java.util.Set; import java.util.TreeSet; import java.util.stream.Collectors; +import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.mockito.Mockito; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper; import org.apache.cassandra.exceptions.ReadFailureException; import org.apache.cassandra.gms.VersionedValue; import org.apache.cassandra.io.util.DataOutputBuffer; @@ -51,7 +54,6 @@ import org.apache.cassandra.utils.JsonUtils; import static org.apache.cassandra.locator.ReplicaUtils.full; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -132,6 +134,15 @@ public class IndexStatusManagerTest } } + @BeforeClass + public static void setUpClass() + { + DatabaseDescriptor.daemonInitialization(); + ClusterMetadataTestHelper.setInstanceForTest(); + for (int i = 251; i <= 255; i++) + ClusterMetadataTestHelper.register(InetAddressAndPort.getByNameUnchecked("127.0.0." + i)); + } + @Test public void shouldPrioritizeSuccessfulEndpoints() { @@ -290,6 +301,7 @@ public class IndexStatusManagerTest @Test public void shouldThrowWhenNotEnoughQueryableEndpoints() { + int port = DatabaseDescriptor.getStoragePort(); assertThatThrownBy(() -> runTest(new Testcase.Builder() .keyspace("ks1") @@ -330,13 +342,14 @@ public class IndexStatusManagerTest .build())) .isInstanceOf(ReadFailureException.class) .hasMessageStartingWith("Operation failed") - .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.252:7000") - .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.254:7000"); + .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.252:" + port) + .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.254:" + port); } @Test public void shouldThrowWhenNoQueryableEndpoints() { + int port = DatabaseDescriptor.getStoragePort(); assertThatThrownBy(() -> runTest(new Testcase.Builder() .keyspace("ks1") @@ -365,9 +378,9 @@ public class IndexStatusManagerTest .build())) .isInstanceOf(ReadFailureException.class) .hasMessageStartingWith("Operation failed") - .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.253:7000") - .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.254:7000") - .hasMessageContaining("INDEX_BUILD_IN_PROGRESS from /127.0.0.255:7000"); + .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.253:" + port) + .hasMessageContaining("INDEX_NOT_AVAILABLE from /127.0.0.254:" + port) + .hasMessageContaining("INDEX_BUILD_IN_PROGRESS from /127.0.0.255:" + port); } void runTest(Testcase testcase) @@ -487,42 +500,4 @@ public class IndexStatusManagerTest assertFalse(IndexStatusManager.shouldWriteToIndexTablesForTesting(new CassandraVersion("5.0.2"))); assertFalse(IndexStatusManager.shouldWriteToIndexTablesForTesting(new CassandraVersion("4.1.0"))); } - - @Test - public void testProcessEventsUpdatesPeerIndexStatus() - { - IndexStatusManager manager = IndexStatusManager.instance; - InetAddressAndPort peer = InetAddressAndPort.getByNameUnchecked("127.0.0.100"); - - Map peerStatuses = new HashMap<>(); - manager.peerIndexStatus.put(peer, peerStatuses); - - assertEquals(Index.Status.UNKNOWN, manager.getIndexStatus(peer, "ks1", "idx1")); - peerStatuses.put("ks1.idx1", Index.Status.BUILD_SUCCEEDED); - assertEquals(Index.Status.BUILD_SUCCEEDED, manager.getIndexStatus(peer, "ks1", "idx1")); - } - - @Test - public void testProcessEventsHandlesDropped() - { - IndexStatusManager manager = IndexStatusManager.instance; - InetAddressAndPort peer = InetAddressAndPort.getByNameUnchecked("127.0.0.101"); - - Map peerStatuses = new HashMap<>(); - peerStatuses.put("ks1.idx1", Index.Status.BUILD_SUCCEEDED); - manager.peerIndexStatus.put(peer, peerStatuses); - - assertEquals(Index.Status.BUILD_SUCCEEDED, manager.getIndexStatus(peer, "ks1", "idx1")); - - peerStatuses.remove("ks1.idx1"); - - assertEquals(Index.Status.UNKNOWN, manager.getIndexStatus(peer, "ks1", "idx1")); - } - - @Test - public void testResetLastPollTimestamp() - { - IndexStatusManager manager = IndexStatusManager.instance; - manager.resetLastPollTimestamp(); - } }