[GitHub] [cassandra] jonmeredith opened a new pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith opened a new pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111


   See https://issues.apache.org/jira/browse/CASSANDRA-16808


-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672637862



##########
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.




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672640517



##########
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672617268



##########
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?




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672681216



##########
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672637077



##########
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.)




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672653081



##########
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
   




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672634341



##########
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.




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672670833



##########
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.
   




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] asfgit closed pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

asfgit closed pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111


   


-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#issuecomment-883727250


   see https://github.com/apache/cassandra/pull/1112/files


-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

maedhroz commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672616626



##########
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.




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672646842



##########
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.




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] jonmeredith commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

jonmeredith commented on a change in pull request #1111:
URL: https://github.com/apache/cassandra/pull/1111#discussion_r672652730



##########
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.




-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

[GitHub] [cassandra] maedhroz commented on a change in pull request #1111: CASSANDRA-16808

[GitBox <gi...@apache.org>]

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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org