HCD-109 Deserialise dse 6.8 PeerInfo files

InetAddressAndPort fields were previously InetAddress, and serialised as either byte[] addresses or text hostnames.

Adds two unit tests, simple serdes of InetAddress to InetAddressAndPort, and parsing of DSE-6 files (kept in `test/data/legacy-metadata/dse-6.8/nodes`).

Author: michaelsembwever

src/java/org/apache/cassandra/nodes/PeerInfo.java

@@ -20,10 +20,12 @@
 import java.util.Objects;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.utils.Throwables;
 
+@JsonIgnoreProperties(ignoreUnknown = true)
 public final class PeerInfo extends NodeInfo
 {
     private final InetAddressAndPort peer;

src/java/org/apache/cassandra/nodes/SerHelper.java

@@ -24,24 +24,29 @@
 import java.util.UUID;
 
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonDeserializer;
 import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.Module;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.commitlog.CommitLogPosition;
 import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.io.util.DataInputBuffer;
-import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.CassandraVersion;
 
 final class SerHelper
 {
+    private static final Logger logger = LoggerFactory.getLogger(SerHelper.class);
+
     static Module createMsgpackModule()
     {
         SimpleModule module = new SimpleModule();
@@ -110,18 +115,34 @@ public void serialize(InetAddressAndPort t, JsonGenerator jsonGenerator, Seriali
         }
     }
 
-    private static final class InetAddressAndPortDeserializer extends JsonDeserializer<InetAddressAndPort>
+    @VisibleForTesting
+    static final class InetAddressAndPortDeserializer extends JsonDeserializer<InetAddressAndPort>
     {
         @Override
         public InetAddressAndPort deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException
         {
-            jsonParser.isExpectedStartArrayToken();
-            jsonParser.nextToken();
-            InetAddress address = InetAddress.getByAddress(jsonParser.getBinaryValue());
-            jsonParser.nextToken();
-            int port = jsonParser.getIntValue();
-            jsonParser.nextToken();
-            return InetAddressAndPort.getByAddressOverrideDefaults(address, port);
+            if (jsonParser.isExpectedStartArrayToken())
+            {
+                jsonParser.nextToken();
+                InetAddress address = InetAddress.getByAddress(jsonParser.getBinaryValue());
+                jsonParser.nextToken();
+                int port = jsonParser.getIntValue();
+                jsonParser.nextToken();
+                return InetAddressAndPort.getByAddressOverrideDefaults(address, port);
+            }
+            try
+            {
+                logger.trace("deserializing legacy InetAddress for field {}",
+                             jsonParser.getParsingContext().hasCurrentName() ? jsonParser.getParsingContext().getCurrentName() : "unknown");
+
+                return InetAddressAndPort.getByAddress(jsonParser.getBinaryValue());
+            }
+            catch (JsonParseException e)
+            {
+                // legacy serialisation of InetAddress can also be text hostname, as well as byte[] address (above)
+                jsonParser.nextToken();
+                return InetAddressAndPort.getByName(jsonParser.getText());
+            }
         }
     }
 

test/data/legacy-metadata/dse-6.8/nodes/local

@@ -18,15 +18,22 @@
 
 package org.apache.cassandra.nodes;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.After;
 import org.junit.Before;
@@ -260,6 +267,50 @@ public void testLocalInfoUnknownFieldsAreIgnoredDuringDeserialization() throws I
         assertEquals(clusterName, loadedLocalInfo.getClusterName());
     }
 
+    @Test
+    public void testDse6InetAddresPeerInfoDeserialisation() throws IOException
+    {
+        InetAddressAndPort peer = InetAddressAndPort.getByName("127.0.0.2");
+        InetAddressSerializer ser = new InetAddressSerializer();
+        SerHelper.InetAddressAndPortDeserializer des = new SerHelper.InetAddressAndPortDeserializer();
+
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        JsonGenerator jsonGenerator = new ObjectMapper().getFactory().createGenerator(outputStream);
+        ser.serialize(peer.address, jsonGenerator, null);
+        jsonGenerator.flush();
+
+        JsonParser jsonParser = new ObjectMapper().getFactory().createParser(outputStream.toByteArray());
+        InetAddressAndPort inap = des.deserialize(jsonParser, null);
+        assertEquals(peer, inap);
+    }
+
+    @Test
+    public void testDse6Compat() throws UnknownHostException
+    {
+        Nodes.Instance.unsafeSetup(Paths.get("test/data/legacy-metadata/dse-6.8/nodes"));
+
+        LocalInfo local = Nodes.local().get();
+        assertEquals("local", local.getKey());
+        assertEquals(BootstrapState.COMPLETED, local.getBootstrapState());
+        assertEquals("3.4.5", local.getCqlVersion().toString());
+        assertEquals("66", local.getNativeProtocolVersion());
+        assertEquals("org.apache.cassandra.dht.Murmur3Partitioner", local.getPartitioner());
+        assertEquals("Cassandra", local.getDataCenter());
+        assertEquals("rack1", local.getRack());
+        assertEquals(256, local.getTokens().size());
+
+        Nodes.Peers peers = Nodes.peers();
+        assertEquals(2, peers.stream().count());
+        PeerInfo peer = peers.get(InetAddressAndPort.getByNameOverrideDefaults("127.0.0.2", 7000));
+        assertEquals("4.0.0.6855", peer.getReleaseVersion().toString());
+        assertEquals("Cassandra", peer.getDataCenter());
+        assertEquals("rack1", peer.getRack());
+        peer = peers.get(InetAddressAndPort.getByNameOverrideDefaults("127.0.0.3", 7000));
+        assertEquals("4.0.0.6855", peer.getReleaseVersion().toString());
+        assertEquals("Cassandra", peer.getDataCenter());
+        assertEquals("rack1", peer.getRack());
+    }
+
     static void fakePeer(PeerInfo p)
     {
         int nodeId = p.getPeer().address.getAddress()[3];
@@ -354,4 +405,14 @@ public void setUnsupportedField(String unsupportedField)
             this.unsupportedField = unsupportedField;
         }
     }
+
+    private static final class InetAddressSerializer extends JsonSerializer<InetAddress>
+    {
+        @Override
+        public void serialize(InetAddress t, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException
+        {
+            // same as https://github.com/riptano/bdp/blob/6.8.54/dse-db/src/java/com/datastax/bdp/db/nodes/SerHelper.java#L200
+            jsonGenerator.writeString(t.getHostAddress());
+        }
+    }
 }