From 4c575a7e4847de174723902fa2eb2e23dfcd192a Mon Sep 17 00:00:00 2001 From: Ariel Weisberg Date: Mon, 13 Nov 2023 17:41:38 -0500 Subject: [PATCH] Fix Paxos V2 prepare response serialization Patch by Ariel Weisberg; Reviewed by David Capwell for CASSANDRA-19023 --- .../cassandra/repair/CassandraRepairJob.java | 3 --- .../cassandra/service/paxos/PaxosPrepare.java | 24 ++++++------------- .../disk/v1/InvertedIndexSearcherTest.java | 13 ++-------- 3 files changed, 9 insertions(+), 31 deletions(-) diff --git a/src/java/org/apache/cassandra/repair/CassandraRepairJob.java b/src/java/org/apache/cassandra/repair/CassandraRepairJob.java index 95a55cd4d5..e2d373c587 100644 --- a/src/java/org/apache/cassandra/repair/CassandraRepairJob.java +++ b/src/java/org/apache/cassandra/repair/CassandraRepairJob.java @@ -47,7 +47,6 @@ import org.apache.cassandra.repair.asymmetric.DifferenceHolder; import org.apache.cassandra.repair.asymmetric.HostDifferences; import org.apache.cassandra.repair.asymmetric.PreferedNodeFilter; import org.apache.cassandra.repair.asymmetric.ReduceHelper; -import org.apache.cassandra.repair.state.JobState; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SystemDistributedKeyspace; import org.apache.cassandra.schema.TableMetadata; @@ -76,7 +75,6 @@ public class CassandraRepairJob extends AbstractRepairJob private static final Logger logger = LoggerFactory.getLogger(CassandraRepairJob.class); private final SharedContext ctx; - public final JobState state; private final RepairJobDesc desc; private final RepairSession session; private final RepairParallelism parallelismDegree; @@ -101,7 +99,6 @@ public class CassandraRepairJob extends AbstractRepairJob this.taskExecutor = session.taskExecutor; this.parallelismDegree = session.parallelismDegree; this.desc = new RepairJobDesc(session.state.parentRepairSession, session.getId(), session.state.keyspace, columnFamily, session.state.commonRange.ranges); - this.state = new JobState(ctx.clock(), desc, session.state.commonRange.endpoints); } public long getNowInSeconds() diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java index eb11bf84ac..1cd7da413c 100644 --- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java +++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java @@ -1258,23 +1258,15 @@ public class PaxosPrepare extends PaxosRequestCallback im } } - private static void serializeRejection(DataOutputPlus out, Ballot supersededBy, ConsensusMigratedAt maybeConsenusMigratedAt, int version) throws IOException - { - out.writeByte(0); - supersededBy.serialize(out); - if (version >= MessagingService.VERSION_51) - ConsensusMigratedAt.serializer.serialize(maybeConsenusMigratedAt, out, version); - } - public static class ResponseSerializer implements IVersionedSerializer { public void serialize(Response response, DataOutputPlus out, int version) throws IOException { if (response.isRejected()) { - out.writeByte(0); Rejected rejected = (Rejected) response; - serializeRejection(out, rejected.supersededBy, rejected.maybeConsenusMigratedAt, version); + out.writeByte(0); + rejected.supersededBy.serialize(out); } else { @@ -1296,9 +1288,9 @@ public class PaxosPrepare extends PaxosRequestCallback im Epoch.messageSerializer.serialize(promised.electorateEpoch, out, version); if (promised.outcome == PERMIT_READ) promised.supersededBy.serialize(out); - if (version >= MessagingService.VERSION_51) - ConsensusMigratedAt.serializer.serialize(response.maybeConsenusMigratedAt, out, version); } + if (version >= MessagingService.VERSION_51) + ConsensusMigratedAt.serializer.serialize(response.maybeConsenusMigratedAt, out, version); } public Response deserialize(DataInputPlus in, int version) throws IOException @@ -1334,17 +1326,15 @@ public class PaxosPrepare extends PaxosRequestCallback im public long serializedSize(Response response, int version) { - long size; + long size = 1; //flags if (response.isRejected()) { - size = 1 + Ballot.sizeInBytes(); - + size += Ballot.sizeInBytes(); } else { Permitted permitted = (Permitted) response; - return 1 - + VIntCoding.computeUnsignedVIntSize(permitted.lowBound) + size += VIntCoding.computeUnsignedVIntSize(permitted.lowBound) + (permitted.latestAcceptedButNotCommitted == null ? 0 : Accepted.serializer.serializedSize(permitted.latestAcceptedButNotCommitted, version)) + Committed.serializer.serializedSize(permitted.latestCommitted, version) + (permitted.readResponse == null ? 0 : ReadResponse.serializer.serializedSize(permitted.readResponse, version)) diff --git a/test/unit/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcherTest.java b/test/unit/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcherTest.java index 93045c80d6..e0385a157d 100644 --- a/test/unit/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcherTest.java +++ b/test/unit/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcherTest.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.List; -import org.junit.BeforeClass; import org.junit.Test; import com.carrotsearch.hppc.LongArrayList; @@ -34,18 +33,17 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.index.sai.QueryContext; import org.apache.cassandra.index.sai.SAITester; import org.apache.cassandra.index.sai.StorageAttachedIndex; -import org.apache.cassandra.index.sai.iterators.KeyRangeIterator; -import org.apache.cassandra.index.sai.memory.MemtableTermsIterator; import org.apache.cassandra.index.sai.disk.PrimaryKeyMap; import org.apache.cassandra.index.sai.disk.format.IndexDescriptor; import org.apache.cassandra.index.sai.disk.v1.segment.IndexSegmentSearcher; import org.apache.cassandra.index.sai.disk.v1.segment.LiteralIndexSegmentSearcher; import org.apache.cassandra.index.sai.disk.v1.segment.SegmentMetadata; import org.apache.cassandra.index.sai.disk.v1.trie.LiteralIndexWriter; +import org.apache.cassandra.index.sai.iterators.KeyRangeIterator; +import org.apache.cassandra.index.sai.memory.MemtableTermsIterator; import org.apache.cassandra.index.sai.plan.Expression; import org.apache.cassandra.index.sai.utils.PrimaryKey; import org.apache.cassandra.index.sai.utils.SAIRandomizedTester; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.bytecomparable.ByteComparable; import org.apache.cassandra.utils.bytecomparable.ByteSource; @@ -92,13 +90,6 @@ public class InvertedIndexSearcherTest extends SAIRandomizedTester }; public static final PrimaryKeyMap.Factory TEST_PRIMARY_KEY_MAP_FACTORY = () -> TEST_PRIMARY_KEY_MAP; - @BeforeClass - public static void setupCQLTester() - { - DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance); - StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance); - } - @Test public void testEqQueriesAgainstStringIndex() throws Exception {