maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672613021
##########
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:
nit: newline
##########
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:
nit: This is very similar to the framework in
`MixedModeReplicationTestBase`, although it clearly adds some new points of
configurability. This test is already written, so it probably doesn't make
sense to rework, but I figured I'd mention it at least to share some tribal
knowledge.
##########
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:
nit: Should we test 3.11 at all, or is the serialization apparatus
basically the same in 3.11 as it is in 3.0?
##########
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;
+ }
+ }
+
+ @Override
+ public InetAddressAndPort deserialize(DataInputPlus in, int version)
throws IOException
+ {
+ if (version >= MessagingService.VERSION_40)
+ {
+ int size = in.readByte() & 0xFF;
+ switch (size)
+ {
+ //Address and one port
+ case 6:
+ case 18:
+ {
+ byte[] bytes = new byte[size - 2];
+ in.readFully(bytes);
+
+ int port = in.readShort() & 0xFFFF;
+ return
getByAddressOverrideDefaults(InetAddress.getByAddress(bytes), bytes, port);
Review comment:
nit: Block is duplicated in the original serializer.
##########
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
Review comment:
nit: Both this and the comments for `Serializer` above could be in
proper JavaDoc format. (Would allow for links and such, etc.)
##########
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:
@jonmeredith There are certainly chunks of `serialize()` and
`serializedSize()` that are duplicated. (i.e. Only their pre-4.0 behavior
diverges?) It may be nice to push some of this into a base class w/ abstract
`pre40Serialize()` and `pre40SerializedSize()` methods to cut down on surface
area until it all goes away if we ever break support for mixed 3.x <-> x.x
version clusters.
##########
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:
nit: unused?
--
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]
