(Accord) NPE while trying to serialize FoundKnownMap as value is null half the time but unexpected while serializing

patch by David Capwell; reviewed by Blake Eggleston for CASSANDRA-19253

modules/accord

@@ -1 +1 @@
-Subproject commit 5523cfefef163efee53c8cc57595f5b50ea4f363
+Subproject commit 0d8f60f742d443365a50115397ff1f0ab10fc694

src/java/org/apache/cassandra/db/Mutation.java

@@ -671,8 +671,12 @@ public class Mutation implements IMutation, Supplier<Mutation>
 
         public PartitionUpdateCollector add(PartitionUpdate partitionUpdate)
         {
-            assert partitionUpdate != null;
+            assert partitionUpdate != null : "Null updates are not allowed";
-            assert partitionUpdate.partitionKey().getPartitioner() == key.getPartitioner();
+            assert partitionUpdate.partitionKey().getPartitioner() == key.getPartitioner(): String.format("Update to key %s with partitioner %s (%s) had an update (%s) with a different partitioner! %s (%s)",
+                                                                                                          key,
+                                                                                                          key.getPartitioner(), key.getPartitioner().getClass(),
+                                                                                                          partitionUpdate,
+                                                                                                          partitionUpdate.partitionKey().getPartitioner(), partitionUpdate.partitionKey().getPartitioner().getClass());
             // note that ImmutableMap.Builder only allows put:ing the same key once, it will fail during build() below otherwise
             modifications.put(partitionUpdate.metadata().id, partitionUpdate);
             empty = false;

src/java/org/apache/cassandra/service/accord/IAccordService.java

@@ -133,7 +133,7 @@ public interface IAccordService
         Set<TableId> allTables = new HashSet<>();
         Set<TableId> newTables = new HashSet<>();
         txn.keys().forEach(key -> {
-            TableId table = ((AccordRoutableKey) key).table();
+            TableId table = key instanceof AccordRoutableKey ? ((AccordRoutableKey) key).table() : ((TokenRange) key).table();
             if (allTables.add(table) && !isAccordManagedTable(table))
                 newTables.add(table);
         });

src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java

@@ -45,6 +45,7 @@ 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.utils.NullableSerializer;
 
 import static accord.messages.CheckStatus.SerializationSupport.createOk;
 
@@ -78,6 +79,8 @@ public class CheckStatusSerializers
         }
     };
 
+    public static final IVersionedSerializer<FoundKnown> foundKnownNullable = NullableSerializer.wrap(foundKnown);
+
     public static final IVersionedSerializer<FoundKnownMap> foundKnownMap = new IVersionedSerializer<>()
     {
         @Override
@@ -88,7 +91,7 @@ public class CheckStatusSerializers
             for (int i = 0 ; i <= size ; ++i)
                 KeySerializers.routingKey.serialize(knownMap.startAt(i), out, version);
             for (int i = 0 ; i < size ; ++i)
-                foundKnown.serialize(knownMap.valueAt(i), out, version);
+                foundKnownNullable.serialize(knownMap.valueAt(i), out, version);
         }
 
         @Override
@@ -100,7 +103,7 @@ public class CheckStatusSerializers
                 starts[i] = KeySerializers.routingKey.deserialize(in, version);
             FoundKnown[] values = new FoundKnown[size];
             for (int i = 0 ; i < size ; ++i)
-                values[i] = foundKnown.deserialize(in, version);
+                values[i] = foundKnownNullable.deserialize(in, version);
             return FoundKnownMap.SerializerSupport.create(true, starts, values);
         }
 
@@ -112,7 +115,7 @@ public class CheckStatusSerializers
             for (int i = 0 ; i <= size ; ++i)
                 result += KeySerializers.routingKey.serializedSize(knownMap.startAt(i), version);
             for (int i = 0 ; i < size ; ++i)
-                result += foundKnown.serializedSize(knownMap.valueAt(i), version);
+                result += foundKnownNullable.serializedSize(knownMap.valueAt(i), version);
             return result;
         }
     };

src/java/org/apache/cassandra/tcm/RemoteProcessor.java

@@ -93,7 +93,6 @@ public final class RemoteProcessor implements Processor
             {
                 log.waitForHighestConsecutive();
             }
-
             return result;
         }
         catch (Exception e)

test/distributed/org/apache/cassandra/distributed/test/tcm/AccordAddTableTest.java

@@ -0,0 +1,80 @@
+/*
+ * 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.test.tcm;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+
+import accord.primitives.Ranges;
+import accord.primitives.Txn;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class AccordAddTableTest extends TestBaseImpl
+{
+    @Test
+    public void test() throws IOException
+    {
+        try (Cluster cluster = builder().withNodes(6)
+                                        .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
+                                        .start())
+        {
+            List<Future<?>> results = new ArrayList<>(cluster.size());
+            for (IInvokableInstance inst : cluster)
+            {
+                Future<?> result = inst.asyncRunsOnInstance(() -> {
+                    for (int i = 0; i < 100; i++)
+                    {
+                        AccordService.instance().maybeConvertTablesToAccord(fakeTxn(i));
+                        if (!ClusterMetadata.current().accordTables.contains(fromNum(i)))
+                            throw new AssertionError("Table not found in TCM!");
+                    }
+                }).call();
+                results.add(result);
+            }
+            FBUtilities.waitOnFutures(results);
+        }
+    }
+
+    private static Txn fakeTxn(int i)
+    {
+        TableId id = fromNum(i);
+
+        Ranges of = Ranges.of(new TokenRange(AccordRoutingKey.SentinelKey.min(id), AccordRoutingKey.SentinelKey.max(id)));
+        return new Txn.InMemory(of, null, null);
+    }
+
+    private static TableId fromNum(int i)
+    {
+        return TableId.fromUUID(new UUID(i, 0)); // not valid... but do we care?
+    }
+}

test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.service.accord.serializers;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
+
+import org.junit.Test;
+
+import accord.api.RoutingKey;
+import accord.coordinate.Infer;
+import accord.local.SaveStatus;
+import accord.messages.CheckStatus.FoundKnownMap;
+import accord.primitives.Ballot;
+import accord.primitives.FullKeyRoute;
+import accord.primitives.Routable;
+import accord.primitives.Unseekables;
+import accord.utils.AccordGens;
+import accord.utils.Gen;
+import accord.utils.Gens;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.AccordGenerators;
+import org.apache.cassandra.utils.CassandraGenerators;
+import org.assertj.core.api.Assertions;
+
+import static accord.utils.Property.qt;
+import static org.apache.cassandra.utils.AccordGenerators.fromQT;
+
+public class CheckStatusSerializersTest
+{
+    static
+    {
+        DatabaseDescriptor.clientInitialization();
+        DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+    }
+
+    @Test
+    public void serde()
+    {
+        DataOutputBuffer buffer = new DataOutputBuffer();
+        qt().forAll(foundKnownMap()).check(map -> Assertions.assertThat(serde(CheckStatusSerializers.foundKnownMap, MessagingService.Version.CURRENT.value, buffer, map)).isEqualTo(map));
+    }
+
+    private static <T> T serde(IVersionedSerializer<T> serializer, int version, DataOutputBuffer buffer, T value) throws IOException
+    {
+        buffer.clear();
+        long expectedSize = serializer.serializedSize(value, version);
+        serializer.serialize(value, buffer, version);
+        Assertions.assertThat(buffer.getLength()).isEqualTo(expectedSize);
+        try (DataInputBuffer in = new DataInputBuffer(buffer.unsafeGetBufferAndFlip(), false))
+        {
+            return serializer.deserialize(in, version);
+        }
+    }
+
+    private static Gen<FoundKnownMap> foundKnownMap()
+    {
+        return rs -> {
+            SaveStatus saveStatus = Gens.pick(SaveStatus.values()).next(rs);
+            Infer.InvalidIfNot invalidIfNot = Gens.pick(Infer.InvalidIfNot.values()).next(rs);
+            Ballot promised = AccordGens.ballot().next(rs);
+            Routable.Domain domain = Gens.pick(Routable.Domain.values()).next(rs);
+            Unseekables<?> keysOrRanges;
+            switch (domain)
+            {
+                case Key:
+                    // TODO (coverage): don't hard code murmur
+                    Gen<AccordRoutingKey> keyGen = AccordGenerators.routingKeyGen(fromQT(CassandraGenerators.TABLE_ID_GEN), Gens.constant(AccordRoutingKey.RoutingKeyKind.TOKEN), fromQT(CassandraGenerators.murmurToken()));
+                    AccordRoutingKey homeKey = keyGen.next(rs);
+                    List<AccordRoutingKey> forOrdering = Gens.lists(keyGen).unique().ofSizeBetween(1, 10).next(rs);
+                    forOrdering.sort(Comparator.naturalOrder());
+                    // TODO (coverage): don't hard code keys type
+                    keysOrRanges = new FullKeyRoute(homeKey, forOrdering.contains(homeKey), forOrdering.toArray(RoutingKey[]::new));
+                    break;
+                case Range:
+                    keysOrRanges = AccordGenerators.ranges(Murmur3Partitioner.instance).next(rs);
+                    break;
+                default:
+                    throw new AssertionError("Unknown domain");
+            }
+            return FoundKnownMap.create(keysOrRanges, saveStatus, invalidIfNot, promised);
+        };
+    }
+}

test/unit/org/apache/cassandra/utils/AccordGenerators.java

@@ -116,11 +116,24 @@ public class AccordGenerators
     }
 
     public static Gen<AccordRoutingKey> routingKeyGen(Gen<TableId> tableIdGen, Gen<Token> tokenGen)
+    {
+        return routingKeyGen(tableIdGen, Gens.enums().all(AccordRoutingKey.RoutingKeyKind.class), tokenGen);
+    }
+
+    public static Gen<AccordRoutingKey> routingKeyGen(Gen<TableId> tableIdGen, Gen<AccordRoutingKey.RoutingKeyKind> kindGen, Gen<Token> tokenGen)
     {
         return rs -> {
             TableId tableId = tableIdGen.next(rs);
-            if (rs.nextBoolean()) return new AccordRoutingKey.TokenKey(tableId, tokenGen.next(rs));
+            AccordRoutingKey.RoutingKeyKind kind = kindGen.next(rs);
-            else return rs.nextBoolean() ? AccordRoutingKey.SentinelKey.min(tableId) : AccordRoutingKey.SentinelKey.max(tableId);
+            switch (kind)
+            {
+                case TOKEN:
+                    return new AccordRoutingKey.TokenKey(tableId, tokenGen.next(rs));
+                case SENTINEL:
+                    return rs.nextBoolean() ? AccordRoutingKey.SentinelKey.min(tableId) : AccordRoutingKey.SentinelKey.max(tableId);
+                default:
+                    throw new AssertionError("Unknown kind: " + kind);
+            }
         };
     }