From a95f072e3527dbb1c8e2d827560b8bdd7939d8fb Mon Sep 17 00:00:00 2001 From: David Capwell Date: Mon, 8 Jan 2024 12:05:54 -0800 Subject: [PATCH] (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 | 2 +- .../org/apache/cassandra/db/Mutation.java | 8 +- .../service/accord/IAccordService.java | 2 +- .../serializers/CheckStatusSerializers.java | 9 +- .../apache/cassandra/tcm/RemoteProcessor.java | 1 - .../test/tcm/AccordAddTableTest.java | 80 +++++++++++++ .../CheckStatusSerializersTest.java | 107 ++++++++++++++++++ .../cassandra/utils/AccordGenerators.java | 17 ++- 8 files changed, 216 insertions(+), 10 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/tcm/AccordAddTableTest.java create mode 100644 test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java diff --git a/modules/accord b/modules/accord index 5523cfefef..0d8f60f742 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 5523cfefef163efee53c8cc57595f5b50ea4f363 +Subproject commit 0d8f60f742d443365a50115397ff1f0ab10fc694 diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index 9c51cd1376..d8ded1eeb1 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -671,8 +671,12 @@ public class Mutation implements IMutation, Supplier public PartitionUpdateCollector add(PartitionUpdate partitionUpdate) { - assert partitionUpdate != null; - assert partitionUpdate.partitionKey().getPartitioner() == key.getPartitioner(); + assert partitionUpdate != null : "Null updates are not allowed"; + 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; diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index e0025b46a3..2f0d7afc71 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -133,7 +133,7 @@ public interface IAccordService Set allTables = new HashSet<>(); Set 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); }); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java index 1d903025af..d4cf093d12 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java +++ b/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 foundKnownNullable = NullableSerializer.wrap(foundKnown); + public static final IVersionedSerializer 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; } }; diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index e5cb0568fe..54adbafba6 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/tcm/AccordAddTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/tcm/AccordAddTableTest.java new file mode 100644 index 0000000000..c486da8f54 --- /dev/null +++ b/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> 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? + } +} diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CheckStatusSerializersTest.java new file mode 100644 index 0000000000..d5da34c8fb --- /dev/null +++ b/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 serde(IVersionedSerializer 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() + { + 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 keyGen = AccordGenerators.routingKeyGen(fromQT(CassandraGenerators.TABLE_ID_GEN), Gens.constant(AccordRoutingKey.RoutingKeyKind.TOKEN), fromQT(CassandraGenerators.murmurToken())); + AccordRoutingKey homeKey = keyGen.next(rs); + List 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); + }; + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index fdd20b0739..4740ee6dfd 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -116,11 +116,24 @@ public class AccordGenerators } public static Gen routingKeyGen(Gen tableIdGen, Gen tokenGen) + { + return routingKeyGen(tableIdGen, Gens.enums().all(AccordRoutingKey.RoutingKeyKind.class), tokenGen); + } + + public static Gen routingKeyGen(Gen tableIdGen, Gen kindGen, Gen tokenGen) { return rs -> { TableId tableId = tableIdGen.next(rs); - if (rs.nextBoolean()) return new AccordRoutingKey.TokenKey(tableId, tokenGen.next(rs)); - else return rs.nextBoolean() ? AccordRoutingKey.SentinelKey.min(tableId) : AccordRoutingKey.SentinelKey.max(tableId); + AccordRoutingKey.RoutingKeyKind kind = kindGen.next(rs); + 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); + } }; }