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

Address review feedback for CASSANDRA-21264

This commit is contained in:
Sunil Ramchandra Pawar 2026-06-30 16:03:27 +05:30
parent e210b54a81
commit 958f4e49b9
4 changed files with 176 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
src/java/org/apache/cassandra/index/IndexStatusManager.java
View File

@ -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<InetAddressAndPort, Map<String, Index.Status>> peerIndexStatus = new HashMap<>();
public final Map<NodeId, Map<String, Index.Status>> 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<String, Index.Status> indexStatusMap;
if (versionedValue.value.equals(TABLE_FALLBACK_MARKER))
{
indexStatusMap = SystemDistributedKeyspace.allIndexStatusesForHost(StorageService.instance.getHostIdForEndpoint(endpoint));
}
else
{
indexStatusMap = statusMapFromString(versionedValue);
}
Map<String, Index.Status> oldStatus = peerIndexStatus.put(endpoint, indexStatusMap);
indexStatusMap = statusMapFromString(versionedValue);
Map<String, Index.Status> oldStatus = peerIndexStatus.put(nodeId, indexStatusMap);
Map<String, Index.Status> updated = updatedIndexStatuses(oldStatus, indexStatusMap);
Set<String> removed = removedIndexStatuses(oldStatus, indexStatusMap);
if (!updated.isEmpty() || !removed.isEmpty())
@ -254,40 +252,36 @@ public class IndexStatusManager
{
try
{
Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
NodeId localNodeId = ClusterMetadata.current().myNodeId();
if (localNodeId == null)
return;
Map<String, Index.Status> 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.
* <p>
@ -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<UUID, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
for (Map.Entry<UUID, Map<String, Index.Status>> entry : allStatuses.entrySet())
Map<NodeId, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
for (Map.Entry<NodeId, Map<String, Index.Status>> 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<UUID, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
for (Map.Entry<UUID, Map<String, Index.Status>> entry : allStatuses.entrySet())
Map<NodeId, Map<String, Index.Status>> allStatuses = SystemDistributedKeyspace.allIndexStatuses();
for (Map.Entry<NodeId, Map<String, Index.Status>> 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<String, Index.Status> statusMap = peerIndexStatus.get(endpoint);
Map<String, Index.Status> 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;
}
}

122
src/java/org/apache/cassandra/schema/SystemDistributedKeyspace.java
View File

@ -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<UUID, Map<String, Index.Status>> allIndexStatuses()
public static Map<NodeId, Map<String, Index.Status>> 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<UUID, Map<String, Index.Status>> allStatuses = new HashMap<>();
Map<NodeId, Map<String, Index.Status>> 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<String, Index.Status> allIndexStatusesForHost(UUID hostId)
public static Map<String, Index.Status> 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;
}
/**

17
test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
View File

@ -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<String, Index.Status> 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<java.util.UUID, java.util.Map<String, Index.Status>> allStatuses =
SystemDistributedKeyspace.allIndexStatuses();
Map<NodeId, Map<String, Index.Status>> 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<java.util.UUID, java.util.Map<String, Index.Status>> allStatuses =
SystemDistributedKeyspace.allIndexStatuses();
Map<NodeId, Map<String, Index.Status>> allStatuses =
SystemDistributedKeyspace.allIndexStatuses();
assertTrue("index_build_status table should still be empty in mixed-version cluster",
allStatuses.isEmpty());
});

63
test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
View File

@ -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<String, Index.Status> 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<String, Index.Status> 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();
}
}