You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/11/26 13:31:04 UTC
(cassandra) 02/02: Merge branch 'cassandra-5.0' into trunk
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 213cdf32fc21e363a88b676f0d7c8a171245793d
Author: Caleb Rackliffe <ca...@gmail.com>
AuthorDate: Thu Nov 21 15:14:17 2024 -0600
Merge branch 'cassandra-5.0' into trunk
* cassandra-5.0:
Streamline the serialized format for index status gossip messages
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/index/Index.java | 36 +++-
.../apache/cassandra/index/IndexStatusManager.java | 118 +++++++++--
.../test/sai/IndexAvailabilityTest.java | 225 ++++++++++++++-------
.../cassandra/index/IndexStatusManagerTest.java | 28 ++-
5 files changed, 301 insertions(+), 107 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 81ff84bc5f..424105bbf8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -88,6 +88,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
+ * Streamline the serialized format for index status gossip messages (CASSANDRA-20058)
* Make configuration entries in memtable section order-independent (CASSANDRA-19906)
* Add guardrail for enabling usage of VectorType (CASSANDRA-19903)
* Set executable flag for shell scripts in .build directory for source artifact (CASSANDRA-19896)
diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java
index 8abc800e0f..f9e2265631 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -1031,10 +1031,36 @@ public interface Index
*/
enum Status
{
- UNKNOWN,
- FULL_REBUILD_STARTED,
- BUILD_FAILED,
- BUILD_SUCCEEDED,
- DROPPED
+ UNKNOWN(0),
+ FULL_REBUILD_STARTED(1),
+ BUILD_FAILED(2),
+ BUILD_SUCCEEDED(3),
+ DROPPED(4);
+
+ public final int code;
+
+ Status(int code)
+ {
+ this.code = code;
+ }
+
+ static Status fromCode(int code)
+ {
+ switch (code)
+ {
+ case 0:
+ return UNKNOWN;
+ case 1:
+ return FULL_REBUILD_STARTED;
+ case 2:
+ return BUILD_FAILED;
+ case 3:
+ return BUILD_SUCCEEDED;
+ case 4:
+ return DROPPED;
+ }
+
+ throw new IllegalArgumentException("Unrecognized code: " + code);
+ }
}
}
diff --git a/src/java/org/apache/cassandra/index/IndexStatusManager.java b/src/java/org/apache/cassandra/index/IndexStatusManager.java
index 8c120efd8f..a745a5640d 100644
--- a/src/java/org/apache/cassandra/index/IndexStatusManager.java
+++ b/src/java/org/apache/cassandra/index/IndexStatusManager.java
@@ -31,6 +31,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,6 +49,7 @@ import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ExecutorUtils;
+import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JsonUtils;
@@ -76,8 +78,7 @@ public class IndexStatusManager
*/
public final Map<InetAddressAndPort, Map<String, Index.Status>> peerIndexStatus = new HashMap<>();
- private IndexStatusManager()
- {}
+ private IndexStatusManager() {}
/**
* Remove endpoints whose indexes are not queryable for the specified {@link Index.QueryPlan}.
@@ -151,29 +152,57 @@ public class IndexStatusManager
if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
return;
- Map<String, String> peerStatus = JsonUtils.fromJsonMap(versionedValue.value);
- Map<String, Index.Status> indexStatus = new HashMap<>();
+ Map<String, Index.Status> indexStatusMap = statusMapFromString(versionedValue);
- for (Map.Entry<String, String> e : peerStatus.entrySet())
- {
- String keyspaceIndex = e.getKey();
- Index.Status status = Index.Status.valueOf(e.getValue());
- indexStatus.put(keyspaceIndex, status);
- }
-
- Map<String, Index.Status> oldStatus = peerIndexStatus.put(endpoint, indexStatus);
- Map<String, Index.Status> updated = updatedIndexStatuses(oldStatus, indexStatus);
- Set<String> removed = removedIndexStatuses(oldStatus, indexStatus);
+ Map<String, Index.Status> oldStatus = peerIndexStatus.put(endpoint, indexStatusMap);
+ Map<String, Index.Status> updated = updatedIndexStatuses(oldStatus, indexStatusMap);
+ Set<String> removed = removedIndexStatuses(oldStatus, indexStatusMap);
if (!updated.isEmpty() || !removed.isEmpty())
logger.debug("Received index status for peer {}:\n Updated: {}\n Removed: {}",
endpoint, updated, removed);
}
- catch (MarshalException | IllegalArgumentException e)
+ catch (Exception e)
{
- logger.warn("Unable to parse index status: {}", e.getMessage());
+ logger.error("Unable to parse index status: {}", e.getMessage());
}
}
+ private Map<String, Index.Status> statusMapFromString(VersionedValue versionedValue)
+ {
+ Map<String, Object> peerStatus = JsonUtils.fromJsonMap(versionedValue.value);
+ Map<String, Index.Status> indexStatusMap = new HashMap<>();
+
+ for (Map.Entry<String, Object> endpointStatus : peerStatus.entrySet())
+ {
+ String keyspaceOrIndex = endpointStatus.getKey();
+ Object keyspaceOrIndexStatus = endpointStatus.getValue();
+
+ if (keyspaceOrIndexStatus instanceof String)
+ {
+ // This is the legacy format: (fully qualified index name -> enum string)
+ Index.Status status = Index.Status.valueOf(keyspaceOrIndexStatus.toString());
+ indexStatusMap.put(keyspaceOrIndex, status);
+ }
+ else if (keyspaceOrIndexStatus instanceof Map)
+ {
+ // This is the new format. (keyspace -> (index -> numeric enum code))
+ @SuppressWarnings("unchecked")
+ Map<String, Integer> keyspaceIndexStatusMap = (Map<String, Integer>) keyspaceOrIndexStatus;
+
+ for (Map.Entry<String, Integer> indexStatus : keyspaceIndexStatusMap.entrySet())
+ {
+ Index.Status status = Index.Status.fromCode(indexStatus.getValue());
+ indexStatusMap.put(identifier(keyspaceOrIndex, indexStatus.getKey()), status);
+ }
+ }
+ else
+ {
+ throw new MarshalException("Invalid index status format: " + endpointStatus);
+ }
+ }
+ return indexStatusMap;
+ }
+
/**
* Propagate a new index status to the ring. The new index status is added to the current index status map
* and the whole map is sent to the ring as a {@link VersionedValue}.
@@ -186,35 +215,80 @@ public class IndexStatusManager
{
try
{
- Map<String, Index.Status> states = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
+ Map<String, Index.Status> statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(),
k -> new HashMap<>());
String keyspaceIndex = identifier(keyspace, index);
if (status == Index.Status.DROPPED)
- states.remove(keyspaceIndex);
+ statusMap.remove(keyspaceIndex);
else
- states.put(keyspaceIndex, status);
+ statusMap.put(keyspaceIndex, status);
// Don't try and propagate if the gossiper isn't enabled. This is primarily for tests where the
// Gossiper has not been started. If we attempt to propagate when not started an exception is
// logged and this causes a number of dtests to fail.
if (Gossiper.instance.isEnabled())
{
- String newStatus = JsonUtils.JSON_OBJECT_MAPPER.writeValueAsString(states);
+ // 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
+ // unknown or one of those 3 versions, continue to propagate the old format.
+ CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
+
+ String newSerializedStatusMap = shouldWriteLegacyStatusFormat(minVersion) ? JsonUtils.writeAsJsonString(statusMap)
+ : toSerializedFormat(statusMap);
+
statusPropagationExecutor.submit(() -> {
// schedule gossiper update asynchronously to avoid potential deadlock when another thread is holding
// gossiper taskLock.
- VersionedValue value = StorageService.instance.valueFactory.indexStatus(newStatus);
+ VersionedValue value = StorageService.instance.valueFactory.indexStatus(newSerializedStatusMap);
Gossiper.instance.addLocalApplicationState(ApplicationState.INDEX_STATUS, value);
});
}
}
- catch (Throwable e)
+ catch (Exception e)
{
logger.warn("Unable to propagate index status: {}", e.getMessage());
}
}
+ private static boolean shouldWriteLegacyStatusFormat(CassandraVersion minVersion)
+ {
+ return minVersion == null || (minVersion.major == 5 && minVersion.minor == 0 && minVersion.patch < 3);
+ }
+
+ /**
+ * Serializes as a JSON string the status of the indexes in the provided map.
+ * <p>
+ * For example, the map...
+ * <pre>
+ * {
+ * ks1.cf1_idx1=FULL_REBUILD_STARTED,
+ * ks1.cf1_idx2=FULL_REBUILD_STARTED,
+ * system.PaxosUncommittedIndex=BUILD_SUCCEEDED
+ * }
+ * </pre>
+ * ...will be converted to the string...
+ * <pre>
+ * {
+ * "system": {"PaxosUncommittedIndex": 3},
+ * "ks1": {"cf1_idx1": 1, "cf1_idx2": 1}
+ * }
+ * </pre>
+ */
+ public static String toSerializedFormat(Map<String, Index.Status> indexStatusMap)
+ {
+ Map<String, Map<String, Integer>> serialized = new HashMap<>();
+
+ for (Map.Entry<String, Index.Status> e : indexStatusMap.entrySet())
+ {
+ String[] keyspaceAndIndex = e.getKey().split("\\.");
+ serialized.computeIfAbsent(keyspaceAndIndex[0], ignore -> new HashMap<>())
+ .put(keyspaceAndIndex[1], e.getValue().code);
+ }
+
+ return JsonUtils.writeAsJsonString(serialized);
+ }
+
@VisibleForTesting
public synchronized Index.Status getIndexStatus(InetAddressAndPort peer, String keyspace, String index)
{
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 3a9e111bad..53cca2614d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java
@@ -29,6 +29,9 @@ import java.util.concurrent.TimeUnit;
import com.google.common.base.Objects;
import org.junit.Test;
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.test.TestBaseImpl;
@@ -41,6 +44,7 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.FBUtilities;
+import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.test.sai.SAIUtil.waitForIndexQueryable;
@@ -62,88 +66,117 @@ 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))
.start()))
{
- String ks1 = "ks1";
- String ks2 = "ks2";
- String ks3 = "ks3";
- String cf1 = "cf1";
- String index1 = "cf1_idx1";
- String index2 = "cf1_idx2";
-
- keyspaces = Arrays.asList(ks1, ks2, ks3);
- indexesPerKs = Arrays.asList(index1, index2);
-
- // create 1 tables per keyspace, 2 indexes per table. all indexes are queryable
- for (String ks : keyspaces)
- {
- cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2));
- cluster.schemaChange(String.format(CREATE_TABLE, ks, cf1));
- cluster.schemaChange(String.format(CREATE_INDEX, index1, ks, cf1, "v1"));
- cluster.schemaChange(String.format(CREATE_INDEX, index2, ks, cf1, "v2"));
- waitForIndexQueryable(cluster, ks);
- cluster.forEach(node -> {
- expectedNodeIndexQueryability.put(NodeIndex.create(ks, index1, node), Index.Status.BUILD_SUCCEEDED);
- expectedNodeIndexQueryability.put(NodeIndex.create(ks, index2, node), Index.Status.BUILD_SUCCEEDED);
- });
- }
+ verifyIndexStatusPropagation(cluster);
+ }
+ }
- // mark ks1 index1 as non-queryable on node1
- markIndexNonQueryable(cluster.get(1), ks1, cf1, index1);
- // on node2, it observes that node1 ks1.index1 is not queryable
- waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_FAILED);
- // other indexes or keyspaces should not be affected
- assertIndexingStatus(cluster);
-
- // mark ks2 index2 as non-queryable on node2
- markIndexNonQueryable(cluster.get(2), ks2, cf1, index2);
- // on node1, it observes that node2 ks2.index2 is not queryable
- waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.BUILD_FAILED);
- // other indexes or keyspaces should not be affected
- assertIndexingStatus(cluster);
-
- // mark ks1 index1 as queryable on node1
- markIndexQueryable(cluster.get(1), ks1, cf1, index1);
- // on node2, it observes that node1 ks1.index1 is queryable
- waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
- // other indexes or keyspaces should not be affected
- assertIndexingStatus(cluster);
-
- // mark ks2 index2 as indexing on node1
- markIndexBuilding(cluster.get(1), ks2, cf1, index2);
- // on node2, it observes that node1 ks2.index2 is not queryable
- waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
- // other indexes or keyspaces should not be affected
- assertIndexingStatus(cluster);
-
- // drop ks1, ks1 index1/index2 should be non queryable on all nodes
- cluster.schemaChange("DROP KEYSPACE " + ks1);
- expectedNodeIndexQueryability.keySet().forEach(k -> {
- if (k.keyspace.equals(ks1))
- expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
- });
- assertIndexingStatus(cluster);
+ @Test
+ public void verifyIndexStatusPropagationMixedPatchVersion() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP).with(NETWORK))
+ .withInstanceInitializer(MixedPatchVersionHelper::setVersions)
+ .start()))
+ {
+ verifyIndexStatusPropagation(cluster);
+ }
+ }
- // drop ks2 index2, there should be no ks2 index2 status on all node
- cluster.schemaChange("DROP INDEX " + ks2 + "." + index2);
- expectedNodeIndexQueryability.keySet().forEach(k -> {
- if (k.keyspace.equals(ks2) && k.index.equals(index2))
- expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
- });
- assertIndexingStatus(cluster);
+ @Test
+ public void verifyIndexStatusPropagationMixedMajorVersion() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withConfig(config -> config.with(GOSSIP).with(NETWORK))
+ .withInstanceInitializer(MixedMajorVersionHelper::setVersions)
+ .start()))
+ {
+ verifyIndexStatusPropagation(cluster);
+ }
+ }
- // drop ks3 cf1, there should be no ks3 index1/index2 status
- cluster.schemaChange("DROP TABLE " + ks3 + "." + cf1);
- expectedNodeIndexQueryability.keySet().forEach(k -> {
- if (k.keyspace.equals(ks3))
- expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
+ private void verifyIndexStatusPropagation(Cluster cluster)
+ {
+ String ks1 = "ks1";
+ String ks2 = "ks2";
+ String ks3 = "ks3";
+ String cf1 = "cf1";
+ String index1 = "cf1_idx1";
+ String index2 = "cf1_idx2";
+
+ keyspaces = Arrays.asList(ks1, ks2, ks3);
+ indexesPerKs = Arrays.asList(index1, index2);
+
+ // create 1 tables per keyspace, 2 indexes per table. all indexes are queryable
+ for (String ks : keyspaces)
+ {
+ cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2));
+ cluster.schemaChange(String.format(CREATE_TABLE, ks, cf1));
+ cluster.schemaChange(String.format(CREATE_INDEX, index1, ks, cf1, "v1"));
+ cluster.schemaChange(String.format(CREATE_INDEX, index2, ks, cf1, "v2"));
+ waitForIndexQueryable(cluster, ks);
+ cluster.forEach(node -> {
+ expectedNodeIndexQueryability.put(NodeIndex.create(ks, index1, node), Index.Status.BUILD_SUCCEEDED);
+ expectedNodeIndexQueryability.put(NodeIndex.create(ks, index2, node), Index.Status.BUILD_SUCCEEDED);
});
- assertIndexingStatus(cluster);
}
+
+ // mark ks1 index1 as non-queryable on node1
+ markIndexNonQueryable(cluster.get(1), ks1, cf1, index1);
+ // on node2, it observes that node1 ks1.index1 is not queryable
+ waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_FAILED);
+ // other indexes or keyspaces should not be affected
+ assertIndexingStatus(cluster);
+
+ // mark ks2 index2 as non-queryable on node2
+ markIndexNonQueryable(cluster.get(2), ks2, cf1, index2);
+ // on node1, it observes that node2 ks2.index2 is not queryable
+ waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.BUILD_FAILED);
+ // other indexes or keyspaces should not be affected
+ assertIndexingStatus(cluster);
+
+ // mark ks1 index1 as queryable on node1
+ markIndexQueryable(cluster.get(1), ks1, cf1, index1);
+ // on node2, it observes that node1 ks1.index1 is queryable
+ waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED);
+ // other indexes or keyspaces should not be affected
+ assertIndexingStatus(cluster);
+
+ // mark ks2 index2 as indexing on node1
+ markIndexBuilding(cluster.get(1), ks2, cf1, index2);
+ // on node2, it observes that node1 ks2.index2 is not queryable
+ waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED);
+ // other indexes or keyspaces should not be affected
+ assertIndexingStatus(cluster);
+
+ // drop ks1, ks1 index1/index2 should be non queryable on all nodes
+ cluster.schemaChange("DROP KEYSPACE " + ks1);
+ expectedNodeIndexQueryability.keySet().forEach(k -> {
+ if (k.keyspace.equals(ks1))
+ expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
+ });
+ assertIndexingStatus(cluster);
+
+ // drop ks2 index2, there should be no ks2 index2 status on all node
+ cluster.schemaChange("DROP INDEX " + ks2 + '.' + index2);
+ expectedNodeIndexQueryability.keySet().forEach(k -> {
+ if (k.keyspace.equals(ks2) && k.index.equals(index2))
+ expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
+ });
+ assertIndexingStatus(cluster);
+
+ // drop ks3 cf1, there should be no ks3 index1/index2 status
+ cluster.schemaChange("DROP TABLE " + ks3 + '.' + cf1);
+ expectedNodeIndexQueryability.keySet().forEach(k -> {
+ if (k.keyspace.equals(ks3))
+ expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN);
+ });
+ assertIndexingStatus(cluster);
}
+ @SuppressWarnings("DataFlowIssue")
private void markIndexNonQueryable(IInvokableInstance node, String keyspace, String table, String indexName)
{
expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_FAILED);
@@ -155,6 +188,7 @@ public class IndexAvailabilityTest extends TestBaseImpl
});
}
+ @SuppressWarnings("DataFlowIssue")
private void markIndexQueryable(IInvokableInstance node, String keyspace, String table, String indexName)
{
expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_SUCCEEDED);
@@ -166,6 +200,7 @@ public class IndexAvailabilityTest extends TestBaseImpl
});
}
+ @SuppressWarnings("DataFlowIssue")
private void markIndexBuilding(IInvokableInstance node, String keyspace, String table, String indexName)
{
expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.FULL_REBUILD_STARTED);
@@ -229,7 +264,7 @@ public class IndexAvailabilityTest extends TestBaseImpl
{
return Index.Status.values()[node.callsOnInstance(() -> getIndexStatus(keyspaceName, indexName, replica).ordinal()).call()];
}
-
+
private static Index.Status getIndexStatus(String keyspaceName, String indexName, InetAddressAndPort replica)
{
KeyspaceMetadata keyspace = Schema.instance.getKeyspaceMetadata(keyspaceName);
@@ -249,7 +284,7 @@ public class IndexAvailabilityTest extends TestBaseImpl
int port = node.callOnInstance(() -> FBUtilities.getBroadcastAddressAndPort().getPort());
return InetAddressAndPort.getByAddressOverrideDefaults(address, port);
}
-
+
private static class NodeIndex
{
private final String keyspace;
@@ -285,4 +320,44 @@ public class IndexAvailabilityTest extends TestBaseImpl
return Objects.hashCode(keyspace, index, node);
}
}
+
+ public static class MixedMajorVersionHelper
+ {
+ @SuppressWarnings({ "unused", "resource" })
+ static void setVersions(ClassLoader loader, int node)
+ {
+ if (node == 1)
+ new ByteBuddy().rebase(FBUtilities.class)
+ .method(named("getReleaseVersionString"))
+ .intercept(MethodDelegation.to(MixedMajorVersionHelper.class))
+ .make()
+ .load(loader, ClassLoadingStrategy.Default.INJECTION);
+ }
+
+ @SuppressWarnings("unused")
+ public static String getReleaseVersionString()
+ {
+ return "4.1.0";
+ }
+ }
+
+ public static class MixedPatchVersionHelper
+ {
+ @SuppressWarnings({ "unused", "resource" })
+ static void setVersions(ClassLoader loader, int node)
+ {
+ if (node == 1)
+ new ByteBuddy().rebase(FBUtilities.class)
+ .method(named("getReleaseVersionString"))
+ .intercept(MethodDelegation.to(MixedPatchVersionHelper.class))
+ .make()
+ .load(loader, ClassLoadingStrategy.Default.INJECTION);
+ }
+
+ @SuppressWarnings("unused")
+ public static String getReleaseVersionString()
+ {
+ return "5.0.2";
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
index 368bd7ccbe..947b7a57bc 100644
--- a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
+++ b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java
@@ -18,12 +18,16 @@
package org.apache.cassandra.index;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.mockito.Mockito;
import org.apache.cassandra.db.ConsistencyLevel;
@@ -43,8 +47,21 @@ import static org.apache.cassandra.locator.ReplicaUtils.full;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertArrayEquals;
+@RunWith(Parameterized.class)
public class IndexStatusManagerTest
{
+ @Parameterized.Parameter
+ public boolean legacyStatusFormat;
+
+ @Parameterized.Parameters(name = "{0}")
+ public static List<Object[]> parameters()
+ {
+ List<Object[]> parameters = new ArrayList<>();
+ parameters.add(new Object[] { true });
+ parameters.add(new Object[] { false });
+ return parameters;
+ }
+
static class Testcase
{
String keyspace;
@@ -362,11 +379,12 @@ public class IndexStatusManagerTest
.collect(Collectors.toSet());
// send indexStatus for each endpoint
- testcase.indexStatus.forEach((endpoint, indexStatus) ->
- IndexStatusManager.instance.receivePeerIndexStatus(
- endpoint,
- VersionedValue.unsafeMakeVersionedValue(JsonUtils.writeAsJsonString(indexStatus), 1)
- ));
+ testcase.indexStatus.forEach((endpoint, indexStatusMap) ->
+ {
+ String serialized = legacyStatusFormat ? JsonUtils.writeAsJsonString(indexStatusMap)
+ : IndexStatusManager.toSerializedFormat(indexStatusMap);
+ IndexStatusManager.instance.receivePeerIndexStatus(endpoint, VersionedValue.unsafeMakeVersionedValue(serialized, 1));
+ });
// sort the replicas here, so that we can assert the order later
EndpointsForRange endpoints = EndpointsForRange.copyOf(new TreeSet<>(replicas));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org