HCD-153 C* 3.x don't like the trailing sha in the version over gossip (VersionedValue)

Author: michaelsembwever

src/java/org/apache/cassandra/gms/EndpointState.java

@@ -24,13 +24,15 @@
 
 import javax.annotation.Nullable;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.CassandraVersion;
 
 /**
@@ -265,7 +267,7 @@ public void serialize(EndpointState epState, DataOutputPlus out, int version) th
         out.writeInt(states.size());
         for (Map.Entry<ApplicationState, VersionedValue> state : states)
         {
-            VersionedValue value = state.getValue();
+            VersionedValue value = filterValue(state.getKey(), state.getValue(), version);
             out.writeInt(state.getKey().ordinal());
             VersionedValue.serializer.serialize(value, out, version);
         }
@@ -294,10 +296,19 @@ public long serializedSize(EndpointState epState, int version)
         size += TypeSizes.sizeof(states.size());
         for (Map.Entry<ApplicationState, VersionedValue> state : states)
         {
-            VersionedValue value = state.getValue();
+            VersionedValue value = filterValue(state.getKey(), state.getValue(), version);
             size += TypeSizes.sizeof(state.getKey().ordinal());
             size += VersionedValue.serializer.serializedSize(value, version);
         }
         return size;
     }
+
+    @VisibleForTesting
+    static VersionedValue filterValue(ApplicationState state, VersionedValue value, int version)
+    {
+        // CC versions come with a sha suffix that C* 3.x nodes cannot parse
+        return version < MessagingService.VERSION_40 && ApplicationState.RELEASE_VERSION == state
+                ? VersionedValue.unsafeMakeVersionedValue(value.value.replaceFirst("-[0-9a-f]{7,40}$", ""), value.version)
+                : value;
+    }
 }

test/unit/org/apache/cassandra/gms/EndpointStateTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.gms;
 
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.Collections;
@@ -33,7 +34,9 @@
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -166,4 +169,34 @@ public void run()
         assertTrue(values.containsKey(ApplicationState.INTERNAL_IP));
         assertTrue(values.containsKey(ApplicationState.HOST_ID));
     }
+
+    @Test
+    public void testCCReleaseVersion() throws IOException
+    {
+        String versionString = "4.0.11.0-0b982c438bfc";
+        String c3safeVersionString = "4.0.11.0";
+        VersionedValue releaseVersion = valueFactory.releaseVersion(versionString);
+
+        assertEquals(versionString, EndpointStateSerializer.filterValue(ApplicationState.RELEASE_VERSION, releaseVersion, MessagingService.VERSION_40).value);
+        assertEquals(c3safeVersionString, EndpointStateSerializer.filterValue(ApplicationState.RELEASE_VERSION, releaseVersion, MessagingService.VERSION_3014).value);
+
+        HeartBeatState hb = new HeartBeatState(0);
+        EndpointState state = new EndpointState(hb);
+        Map<ApplicationState, VersionedValue> states = new EnumMap<>(ApplicationState.class);
+        states.put(ApplicationState.RELEASE_VERSION, releaseVersion);
+        state.addApplicationStates(states);
+        assertEquals(versionString, state.getReleaseVersion().toString());
+
+        DataOutputBuffer buffer = new DataOutputBuffer();
+        EndpointState.serializer.serialize(state, buffer, MessagingService.VERSION_40);
+        DataInputBuffer input = new DataInputBuffer(buffer.buffer(), false);
+        EndpointState deserializedCurrent = EndpointState.serializer.deserialize(input, MessagingService.VERSION_40);
+        assertEquals(versionString, deserializedCurrent.getApplicationState(ApplicationState.RELEASE_VERSION).value);
+
+        buffer = new DataOutputBuffer();
+        EndpointState.serializer.serialize(state, buffer, MessagingService.VERSION_3014);
+        input = new DataInputBuffer(buffer.buffer(), false);
+        EndpointState deserializedOld = EndpointState.serializer.deserialize(input, MessagingService.VERSION_3014);
+        assertEquals(c3safeVersionString, deserializedOld.getApplicationState(ApplicationState.RELEASE_VERSION).value);
+    }
 }