jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672646306
##########
File path: src/java/org/apache/cassandra/net/Message.java
##########
@@ -1151,9 +1151,20 @@ private void serializeParams(Map<ParamType, Object>
params, DataOutputPlus out,
: in.readInt();
if (null != type)
- params.put(type, type.serializer.deserialize(in, version));
- else
+ {
+ // Have to special case deserializer as pre-4.0 needs
length to decode correctly
+ if (version < VERSION_40 && type == ParamType.RESPOND_TO)
+ {
+ params.put(type,
InetAddressAndPort.FwdFrmSerializer.fwdFrmSerializer.pre40DeserializeWithLength(in,
version, length));
+ }
+ else
+ {
+ params.put(type, type.serializer.deserialize(in,
version));
+ }
+ }
+ else {
Review comment:
ack
##########
File path:
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.Shared;
+import org.apache.cassandra.distributed.shared.Versions;
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.*;
+
+@Shared
+public class MixedModeMessageForwardTest extends UpgradeTestBase
+{
+ private static final Logger logger =
LoggerFactory.getLogger(MixedModeMessageForwardTest.class);
+ private static int nextKey = 1;
+ private static String TABLE = "tbl";
+ private static String INSERT_QUERY = String.format("INSERT INTO %s.%s(pk)
VALUES (?)", KEYSPACE, TABLE);
+ private static String CHECK_QUERY = String.format("SELECT pk FROM %s.%s
WHERE pk = ?", KEYSPACE, TABLE);
+
+ private boolean checkClusterUp(UpgradeableCluster cluster, int coordId)
+ {
+ NodeToolResult result;
+ result = cluster.get(coordId).nodetoolResult("ring");
+
+ // Must have an Up line for each node
+ long upCount =
Arrays.stream(result.getStdout().split("\\r?\\n")).filter(line ->
line.contains(" Up ")).count();
+ if (upCount < cluster.size())
+ {
+ logger.info("Only {}/{} are up.", upCount, cluster.size());
+ return false;
+ }
+
+ return true;
+ }
+
+ private void writeReadTest(UpgradeableCluster cluster)
+ {
+ // Coordinate a write from each node and then check present on all
replicas
+ int readKey = nextKey;
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ final int checkNodeId = coordId;
+ Awaitility.await("Cluster Up").atMost(1,
TimeUnit.MINUTES).until(() -> checkClusterUp(cluster, checkNodeId));
+
+ logger.info("Coordinating CL.ALL Insert from node{} ", coordId);
+ cluster.get(coordId).coordinator().execute(INSERT_QUERY,
ConsistencyLevel.ALL, nextKey++);
+ }
+
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ for (int nodeId = 1; nodeId <= cluster.size(); nodeId++) {
+ Object[][] results =
cluster.get(nodeId).executeInternal(CHECK_QUERY, readKey);
+ assertRows(results, row(readKey));
+ }
+ readKey++;
+ }
+ }
+
+ /* Verify that messages sent with sendToHintedReplicas to non-local DCs
+ * are forwarded on to the hosts there.
+ *
+ * 1) creates a mixed cluster with multiple datacenters and a keyspace
+ * configured to write to all replicas in the datacenter
+ * 2) check the original single-version cluster by issuing an INSERT
+ * mutation from a coordinator on each node, then check that value
+ * has locally been written to each of the nodes.
+ * 3) Upgrade nodes one at a time, rechecking that all writes are
forwarded.
+ */
+ @Test
+ public void checkWritesForwardedToOtherDcTest() throws Throwable
+ {
+ int numDCs = 2;
+ int nodesPerDc = 3;
+ String ntsArgs = IntStream.range(1, numDCs + 1)
+ .mapToObj(dc ->
String.format("'datacenter%d' : %d", dc, nodesPerDc))
+ .collect(Collectors.joining(","));
+
+ new TestCase()
+ .withConfig(c -> c.with(Feature.GOSSIP,
Feature.NETWORK).set("request_timeout_in_ms", 30000))
+ .withBuilder(b -> b.withRacks(numDCs, 1, nodesPerDc))
+ .nodes(numDCs * nodesPerDc)
+ .upgrade(Versions.Major.v30, Versions.Major.v4)
Review comment:
Serialization looked the same when I checked 3.11. Not sure it's worth
running every release, but I can test it once off.
##########
File path: src/java/org/apache/cassandra/locator/InetAddressAndPort.java
##########
@@ -382,4 +385,83 @@ public long serializedSize(InetAddressAndPort from, int
version)
}
}
}
+
+ // Serializer for handling FWD_FRM message parameters. Pre-4.0
deserialization is a special
+ // case in the message
+ public static final class FwdFrmSerializer implements
IVersionedSerializer<InetAddressAndPort>
+ {
+ public static final FwdFrmSerializer fwdFrmSerializer = new
FwdFrmSerializer();
+ private FwdFrmSerializer() { }
+
+ public void serialize(InetAddressAndPort endpoint, DataOutputPlus out,
int version) throws IOException
+ {
+ byte[] buf = endpoint.addressBytes;
+
+ if (version >= MessagingService.VERSION_40)
+ {
+ out.writeByte(buf.length + 2);
+ out.write(buf);
+ out.writeShort(endpoint.port);
+ }
+ else
+ {
+ out.write(buf);
+ }
+ }
+
+ public long serializedSize(InetAddressAndPort from, int version)
+ {
+ //4.0 includes a port number
+ if (version >= MessagingService.VERSION_40)
+ {
+ if (from.address instanceof Inet4Address)
+ return 1 + 4 + 2;
+ assert from.address instanceof Inet6Address;
+ return 1 + 16 + 2;
+ }
+ else
+ {
+ if (from.address instanceof Inet4Address)
+ return 4;
+ assert from.address instanceof Inet6Address;
+ return 16;
+ }
+ }
Review comment:
I had the patch that way originally but backed away as each method has
to handle pre/post 4.0 messages so to when adding protections to make sure the
wrong serializer wasn't called call paths longer/uglier so I went back to the
simpler option of having the two full implementations.
##########
File path: src/java/org/apache/cassandra/locator/InetAddressAndPort.java
##########
@@ -34,6 +34,7 @@
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.ParamType;
Review comment:
ack
##########
File path:
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.Shared;
+import org.apache.cassandra.distributed.shared.Versions;
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.*;
+
+@Shared
+public class MixedModeMessageForwardTest extends UpgradeTestBase
+{
+ private static final Logger logger =
LoggerFactory.getLogger(MixedModeMessageForwardTest.class);
+ private static int nextKey = 1;
+ private static String TABLE = "tbl";
+ private static String INSERT_QUERY = String.format("INSERT INTO %s.%s(pk)
VALUES (?)", KEYSPACE, TABLE);
+ private static String CHECK_QUERY = String.format("SELECT pk FROM %s.%s
WHERE pk = ?", KEYSPACE, TABLE);
+
+ private boolean checkClusterUp(UpgradeableCluster cluster, int coordId)
+ {
+ NodeToolResult result;
+ result = cluster.get(coordId).nodetoolResult("ring");
+
+ // Must have an Up line for each node
+ long upCount =
Arrays.stream(result.getStdout().split("\\r?\\n")).filter(line ->
line.contains(" Up ")).count();
+ if (upCount < cluster.size())
+ {
+ logger.info("Only {}/{} are up.", upCount, cluster.size());
+ return false;
+ }
+
+ return true;
+ }
+
+ private void writeReadTest(UpgradeableCluster cluster)
+ {
+ // Coordinate a write from each node and then check present on all
replicas
+ int readKey = nextKey;
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ final int checkNodeId = coordId;
+ Awaitility.await("Cluster Up").atMost(1,
TimeUnit.MINUTES).until(() -> checkClusterUp(cluster, checkNodeId));
+
+ logger.info("Coordinating CL.ALL Insert from node{} ", coordId);
+ cluster.get(coordId).coordinator().execute(INSERT_QUERY,
ConsistencyLevel.ALL, nextKey++);
+ }
+
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ for (int nodeId = 1; nodeId <= cluster.size(); nodeId++) {
+ Object[][] results =
cluster.get(nodeId).executeInternal(CHECK_QUERY, readKey);
+ assertRows(results, row(readKey));
+ }
+ readKey++;
+ }
+ }
+
+ /* Verify that messages sent with sendToHintedReplicas to non-local DCs
+ * are forwarded on to the hosts there.
+ *
+ * 1) creates a mixed cluster with multiple datacenters and a keyspace
+ * configured to write to all replicas in the datacenter
+ * 2) check the original single-version cluster by issuing an INSERT
+ * mutation from a coordinator on each node, then check that value
+ * has locally been written to each of the nodes.
+ * 3) Upgrade nodes one at a time, rechecking that all writes are
forwarded.
+ */
+ @Test
+ public void checkWritesForwardedToOtherDcTest() throws Throwable
+ {
+ int numDCs = 2;
+ int nodesPerDc = 3;
+ String ntsArgs = IntStream.range(1, numDCs + 1)
+ .mapToObj(dc ->
String.format("'datacenter%d' : %d", dc, nodesPerDc))
+ .collect(Collectors.joining(","));
+
+ new TestCase()
+ .withConfig(c -> c.with(Feature.GOSSIP,
Feature.NETWORK).set("request_timeout_in_ms", 30000))
+ .withBuilder(b -> b.withRacks(numDCs, 1, nodesPerDc))
+ .nodes(numDCs * nodesPerDc)
+ .upgrade(Versions.Major.v30, Versions.Major.v4)
Review comment:
Confirmed running with `.upgrade(Versions.Major.v3X, Versions.Major.v4)`
works.
Before this change there would be value in testing `nodesPerDC=2;` and a
larger value to exercise different paths through `sameMessageId`, however the
updated implementation doesn't special case single node forwarding.
##########
File path:
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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 org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.Shared;
+import org.apache.cassandra.distributed.shared.Versions;
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.*;
+
+@Shared
+public class MixedModeMessageForwardTest extends UpgradeTestBase
+{
+ private static final Logger logger =
LoggerFactory.getLogger(MixedModeMessageForwardTest.class);
+ private static int nextKey = 1;
+ private static String TABLE = "tbl";
+ private static String INSERT_QUERY = String.format("INSERT INTO %s.%s(pk)
VALUES (?)", KEYSPACE, TABLE);
+ private static String CHECK_QUERY = String.format("SELECT pk FROM %s.%s
WHERE pk = ?", KEYSPACE, TABLE);
+
+ private boolean checkClusterUp(UpgradeableCluster cluster, int coordId)
+ {
+ NodeToolResult result;
+ result = cluster.get(coordId).nodetoolResult("ring");
+
+ // Must have an Up line for each node
+ long upCount =
Arrays.stream(result.getStdout().split("\\r?\\n")).filter(line ->
line.contains(" Up ")).count();
+ if (upCount < cluster.size())
+ {
+ logger.info("Only {}/{} are up.", upCount, cluster.size());
+ return false;
+ }
+
+ return true;
+ }
+
+ private void writeReadTest(UpgradeableCluster cluster)
+ {
+ // Coordinate a write from each node and then check present on all
replicas
+ int readKey = nextKey;
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ final int checkNodeId = coordId;
+ Awaitility.await("Cluster Up").atMost(1,
TimeUnit.MINUTES).until(() -> checkClusterUp(cluster, checkNodeId));
+
+ logger.info("Coordinating CL.ALL Insert from node{} ", coordId);
+ cluster.get(coordId).coordinator().execute(INSERT_QUERY,
ConsistencyLevel.ALL, nextKey++);
+ }
+
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ for (int nodeId = 1; nodeId <= cluster.size(); nodeId++) {
+ Object[][] results =
cluster.get(nodeId).executeInternal(CHECK_QUERY, readKey);
+ assertRows(results, row(readKey));
+ }
+ readKey++;
+ }
+ }
+
+ /* Verify that messages sent with sendToHintedReplicas to non-local DCs
+ * are forwarded on to the hosts there.
+ *
+ * 1) creates a mixed cluster with multiple datacenters and a keyspace
+ * configured to write to all replicas in the datacenter
+ * 2) check the original single-version cluster by issuing an INSERT
+ * mutation from a coordinator on each node, then check that value
+ * has locally been written to each of the nodes.
+ * 3) Upgrade nodes one at a time, rechecking that all writes are
forwarded.
+ */
+ @Test
+ public void checkWritesForwardedToOtherDcTest() throws Throwable
+ {
+ int numDCs = 2;
+ int nodesPerDc = 3;
+ String ntsArgs = IntStream.range(1, numDCs + 1)
+ .mapToObj(dc ->
String.format("'datacenter%d' : %d", dc, nodesPerDc))
+ .collect(Collectors.joining(","));
+
+ new TestCase()
+ .withConfig(c -> c.with(Feature.GOSSIP,
Feature.NETWORK).set("request_timeout_in_ms", 30000))
+ .withBuilder(b -> b.withRacks(numDCs, 1, nodesPerDc))
+ .nodes(numDCs * nodesPerDc)
+ .upgrade(Versions.Major.v30, Versions.Major.v4)
+ .setup(cluster -> {
+ cluster.schemaChange("ALTER KEYSPACE " + KEYSPACE +
+ " WITH replication = {'class': 'NetworkTopologyStrategy', " +
ntsArgs + " };");
+
+ cluster.schemaChange("CREATE TABLE "+ KEYSPACE + "." + TABLE + "
(pk int, PRIMARY KEY(pk))");
+
+ logger.info("Testing after setup, all nodes running {}",
cluster.get(1).getReleaseVersionString());
+ writeReadTest(cluster);
+ })
+ .runAfterNodeUpgrade((UpgradeableCluster cluster, int nodeId) -> {
+ // Should be able to coordinate a write to any node and have a
copy appear locally on all others
+ logger.info("Testing after upgrading node{} to {}", nodeId,
cluster.get(nodeId).getReleaseVersionString());
+ writeReadTest(cluster);
+ })
+ .run();
Review comment:
I took a look at unifying them and the sticking point is I don't really
want to re-implement NTS to work out where things should be written.
`MixedModeMessageForwardTest` sets the RF to make sure all nodes in the DC are
written to so it's easier to check.
I was able to remove node UP check, I think I added it when I didn't
understand the messageId issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
