natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Paxos V2 prepare response serialization 

Patch by Ariel Weisberg; Reviewed by David Capwell for CASSANDRA-19023
This commit is contained in:
Ariel Weisberg 2023-11-13 17:41:38 -05:00 committed by David Capwell
parent c10c84b9cd
commit 4c575a7e48
3 changed files with 9 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/repair/CassandraRepairJob.java
View File

@ -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()

24
src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
View File

@ -1258,23 +1258,15 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> 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<Response>
{
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<PaxosPrepare.Response> 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<PaxosPrepare.Response> 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))

13
test/unit/org/apache/cassandra/index/sai/disk/v1/InvertedIndexSearcherTest.java
View File

@ -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
{