mirror of https://github.com/apache/cassandra
749 lines
62 KiB
Java
749 lines
62 KiB
Java
/*
|
|
* 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.net;
|
|
|
|
import java.lang.reflect.Field;
|
|
import java.lang.reflect.Modifier;
|
|
import java.util.List;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.function.Supplier;
|
|
import java.util.function.ToLongFunction;
|
|
|
|
import com.google.common.annotations.VisibleForTesting;
|
|
import com.google.common.collect.ImmutableList;
|
|
|
|
import org.apache.cassandra.batchlog.Batch;
|
|
import org.apache.cassandra.batchlog.BatchRemoveVerbHandler;
|
|
import org.apache.cassandra.batchlog.BatchStoreVerbHandler;
|
|
import org.apache.cassandra.concurrent.Stage;
|
|
import org.apache.cassandra.config.DatabaseDescriptor;
|
|
import org.apache.cassandra.db.CounterMutation;
|
|
import org.apache.cassandra.db.CounterMutationVerbHandler;
|
|
import org.apache.cassandra.db.Mutation;
|
|
import org.apache.cassandra.db.MutationVerbHandler;
|
|
import org.apache.cassandra.db.ReadCommand;
|
|
import org.apache.cassandra.db.ReadCommandVerbHandler;
|
|
import org.apache.cassandra.db.ReadRepairVerbHandler;
|
|
import org.apache.cassandra.db.ReadResponse;
|
|
import org.apache.cassandra.db.SnapshotCommand;
|
|
import org.apache.cassandra.db.TruncateRequest;
|
|
import org.apache.cassandra.db.TruncateResponse;
|
|
import org.apache.cassandra.db.TruncateVerbHandler;
|
|
import org.apache.cassandra.db.compression.CompressionDictionaryUpdateMessage;
|
|
import org.apache.cassandra.db.compression.CompressionDictionaryUpdateVerbHandler;
|
|
import org.apache.cassandra.db.virtual.VirtualMutation;
|
|
import org.apache.cassandra.exceptions.RequestFailure;
|
|
import org.apache.cassandra.gms.GossipDigestAck;
|
|
import org.apache.cassandra.gms.GossipDigestAck2;
|
|
import org.apache.cassandra.gms.GossipDigestAck2VerbHandler;
|
|
import org.apache.cassandra.gms.GossipDigestAckVerbHandler;
|
|
import org.apache.cassandra.gms.GossipDigestSyn;
|
|
import org.apache.cassandra.gms.GossipDigestSynVerbHandler;
|
|
import org.apache.cassandra.gms.GossipShutdown;
|
|
import org.apache.cassandra.gms.GossipShutdownVerbHandler;
|
|
import org.apache.cassandra.hints.HintMessage;
|
|
import org.apache.cassandra.hints.HintVerbHandler;
|
|
import org.apache.cassandra.io.AsymmetricUnversionedSerializer;
|
|
import org.apache.cassandra.io.AsymmetricVersionedSerializer;
|
|
import org.apache.cassandra.io.IVersionedAsymmetricSerializer;
|
|
import org.apache.cassandra.repair.RepairMessageVerbHandler;
|
|
import org.apache.cassandra.repair.messages.CleanupMessage;
|
|
import org.apache.cassandra.repair.messages.FailSession;
|
|
import org.apache.cassandra.repair.messages.FinalizeCommit;
|
|
import org.apache.cassandra.repair.messages.FinalizePromise;
|
|
import org.apache.cassandra.repair.messages.FinalizePropose;
|
|
import org.apache.cassandra.repair.messages.MutationTrackingSyncRequest;
|
|
import org.apache.cassandra.repair.messages.MutationTrackingSyncResponse;
|
|
import org.apache.cassandra.repair.messages.PrepareConsistentRequest;
|
|
import org.apache.cassandra.repair.messages.PrepareConsistentResponse;
|
|
import org.apache.cassandra.repair.messages.PrepareMessage;
|
|
import org.apache.cassandra.repair.messages.SnapshotMessage;
|
|
import org.apache.cassandra.repair.messages.StatusRequest;
|
|
import org.apache.cassandra.repair.messages.StatusResponse;
|
|
import org.apache.cassandra.repair.messages.SyncRequest;
|
|
import org.apache.cassandra.repair.messages.SyncResponse;
|
|
import org.apache.cassandra.repair.messages.ValidationRequest;
|
|
import org.apache.cassandra.repair.messages.ValidationResponse;
|
|
import org.apache.cassandra.replication.ActivationResponse;
|
|
import org.apache.cassandra.replication.BroadcastLogOffsets;
|
|
import org.apache.cassandra.replication.ForwardedWrite;
|
|
import org.apache.cassandra.replication.TransferTrackingService;
|
|
import org.apache.cassandra.replication.PullMutationsRequest;
|
|
import org.apache.cassandra.replication.PushMutationRequest;
|
|
import org.apache.cassandra.replication.ActivationRequest;
|
|
import org.apache.cassandra.replication.TransferFailed;
|
|
import org.apache.cassandra.schema.SchemaMutationsSerializer;
|
|
import org.apache.cassandra.schema.SchemaPullVerbHandler;
|
|
import org.apache.cassandra.schema.SchemaPushVerbHandler;
|
|
import org.apache.cassandra.schema.SchemaVersionVerbHandler;
|
|
import org.apache.cassandra.service.EchoVerbHandler;
|
|
import org.apache.cassandra.service.SnapshotVerbHandler;
|
|
import org.apache.cassandra.service.accord.AccordSerializers;
|
|
import org.apache.cassandra.service.accord.AccordService;
|
|
import org.apache.cassandra.service.accord.AccordSyncPropagator;
|
|
import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification;
|
|
import org.apache.cassandra.service.accord.FetchTopologies;
|
|
import org.apache.cassandra.service.accord.WatermarkCollector;
|
|
import org.apache.cassandra.service.accord.interop.AccordInteropApply;
|
|
import org.apache.cassandra.service.accord.interop.AccordInteropRead;
|
|
import org.apache.cassandra.service.accord.interop.AccordInteropReadRepair;
|
|
import org.apache.cassandra.service.accord.interop.AccordInteropStableThenRead;
|
|
import org.apache.cassandra.service.accord.serializers.AcceptSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.ApplySerializers;
|
|
import org.apache.cassandra.service.accord.serializers.AwaitSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.BeginInvalidationSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.CheckStatusSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.CommitSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.EnumSerializer;
|
|
import org.apache.cassandra.service.accord.serializers.FetchSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.GetDurableBeforeSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.GetEphmrlReadDepsSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.GetMaxConflictSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.InformSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.LatestDepsSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.PreacceptSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.ReadDataSerializer;
|
|
import org.apache.cassandra.service.accord.serializers.RecoverySerializers;
|
|
import org.apache.cassandra.service.accord.serializers.SetDurableSerializers;
|
|
import org.apache.cassandra.service.accord.serializers.Version;
|
|
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState;
|
|
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.ConsensusKeyMigrationFinished;
|
|
import org.apache.cassandra.service.paxos.CasForwardHandler;
|
|
import org.apache.cassandra.service.paxos.CasForwardRequest;
|
|
import org.apache.cassandra.service.paxos.CasForwardResponse;
|
|
import org.apache.cassandra.service.paxos.Commit;
|
|
import org.apache.cassandra.service.paxos.Commit.Agreed;
|
|
import org.apache.cassandra.service.paxos.ConsensusReadForwardHandler;
|
|
import org.apache.cassandra.service.paxos.ConsensusReadForwardRequest;
|
|
import org.apache.cassandra.service.paxos.Paxos2CommitForwardHandler;
|
|
import org.apache.cassandra.service.paxos.Paxos2CommitForwardRequest;
|
|
import org.apache.cassandra.service.paxos.PaxosCommit;
|
|
import org.apache.cassandra.service.paxos.PaxosCommitAndPrepare;
|
|
import org.apache.cassandra.service.paxos.PaxosCommitForwardHandler;
|
|
import org.apache.cassandra.service.paxos.PaxosCommitForwardRequest;
|
|
import org.apache.cassandra.service.paxos.PrepareRefreshForwardHandler;
|
|
import org.apache.cassandra.service.paxos.PrepareRefreshForwardRequest;
|
|
import org.apache.cassandra.service.paxos.PrepareRefreshForwardResponse;
|
|
import org.apache.cassandra.service.paxos.PaxosPrepare;
|
|
import org.apache.cassandra.service.paxos.PaxosPrepareRefresh;
|
|
import org.apache.cassandra.service.paxos.PaxosPropose;
|
|
import org.apache.cassandra.service.paxos.PaxosRepair;
|
|
import org.apache.cassandra.service.paxos.PrepareResponse;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupComplete;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupHistory;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupRequest;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupResponse;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosFinishPrepareCleanup;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosStartPrepareCleanup;
|
|
import org.apache.cassandra.service.paxos.cleanup.PaxosUpdateLowBallot;
|
|
import org.apache.cassandra.service.paxos.v1.PrepareVerbHandler;
|
|
import org.apache.cassandra.service.paxos.v1.ProposeVerbHandler;
|
|
import org.apache.cassandra.service.reads.tracked.ReadReconcileAck;
|
|
import org.apache.cassandra.service.reads.tracked.TrackedDataResponse;
|
|
import org.apache.cassandra.service.reads.tracked.TrackedRead;
|
|
import org.apache.cassandra.service.reads.tracked.TrackedSummaryResponse;
|
|
import org.apache.cassandra.streaming.DataMovement;
|
|
import org.apache.cassandra.streaming.DataMovementVerbHandler;
|
|
import org.apache.cassandra.streaming.ReplicationDoneVerbHandler;
|
|
import org.apache.cassandra.tcm.Discovery;
|
|
import org.apache.cassandra.tcm.Epoch;
|
|
import org.apache.cassandra.tcm.FetchCMSLog;
|
|
import org.apache.cassandra.tcm.FetchPeerLog;
|
|
import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
|
|
import org.apache.cassandra.tcm.migration.CMSInitializationResponse;
|
|
import org.apache.cassandra.tcm.migration.Election;
|
|
import org.apache.cassandra.tcm.sequences.DataMovements;
|
|
import org.apache.cassandra.tcm.serialization.MessageSerializers;
|
|
import org.apache.cassandra.utils.BooleanSerializer;
|
|
import org.apache.cassandra.utils.ReflectionUtils;
|
|
import org.apache.cassandra.utils.TimeUUID;
|
|
import org.apache.cassandra.utils.UUIDSerializer;
|
|
|
|
import static java.util.concurrent.TimeUnit.NANOSECONDS;
|
|
import static org.apache.cassandra.concurrent.Stage.ANTI_ENTROPY;
|
|
import static org.apache.cassandra.concurrent.Stage.COUNTER_MUTATION;
|
|
import static org.apache.cassandra.concurrent.Stage.FETCH_METADATA;
|
|
import static org.apache.cassandra.concurrent.Stage.GOSSIP;
|
|
import static org.apache.cassandra.concurrent.Stage.IMMEDIATE;
|
|
import static org.apache.cassandra.concurrent.Stage.INTERNAL_METADATA;
|
|
import static org.apache.cassandra.concurrent.Stage.INTERNAL_RESPONSE;
|
|
import static org.apache.cassandra.concurrent.Stage.MIGRATION;
|
|
import static org.apache.cassandra.concurrent.Stage.MISC;
|
|
import static org.apache.cassandra.concurrent.Stage.MUTATION;
|
|
import static org.apache.cassandra.concurrent.Stage.PAXOS_REPAIR;
|
|
import static org.apache.cassandra.concurrent.Stage.READ;
|
|
import static org.apache.cassandra.concurrent.Stage.REQUEST_RESPONSE;
|
|
import static org.apache.cassandra.concurrent.Stage.TRACKED_CAS;
|
|
import static org.apache.cassandra.concurrent.Stage.TRACING;
|
|
import static org.apache.cassandra.net.ResponseHandlerSupplier.RESPONSE_HANDLER;
|
|
import static org.apache.cassandra.net.Verb.Kind.CUSTOM;
|
|
import static org.apache.cassandra.net.Verb.Kind.NORMAL;
|
|
import static org.apache.cassandra.net.Verb.Priority.P0;
|
|
import static org.apache.cassandra.net.Verb.Priority.P1;
|
|
import static org.apache.cassandra.net.Verb.Priority.P2;
|
|
import static org.apache.cassandra.net.Verb.Priority.P3;
|
|
import static org.apache.cassandra.net.Verb.Priority.P4;
|
|
import static org.apache.cassandra.net.VerbTimeouts.counterTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.longTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.noTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.pingTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.rangeTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.readTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.repairTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.repairValidationRspTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.repairWithBackoffTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.rpcTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.shortTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.truncateTimeout;
|
|
import static org.apache.cassandra.net.VerbTimeouts.writeTimeout;
|
|
import static org.apache.cassandra.tcm.ClusterMetadataService.commitRequestHandler;
|
|
import static org.apache.cassandra.tcm.ClusterMetadataService.currentEpochRequestHandler;
|
|
import static org.apache.cassandra.tcm.ClusterMetadataService.fetchLogRequestHandler;
|
|
import static org.apache.cassandra.tcm.ClusterMetadataService.logNotifyHandler;
|
|
import static org.apache.cassandra.tcm.ClusterMetadataService.replicationHandler;
|
|
|
|
/**
|
|
* Note that priorities except P0 are presently unused. P0 corresponds to urgent, i.e. what used to be the "Gossip" connection.
|
|
*/
|
|
@SuppressWarnings("Convert2MethodRef") // we must defer all initialisation, which includes e.g. taking a method reference to a static object/singleton, which this inspection does not disambiguate
|
|
public enum Verb
|
|
{
|
|
MUTATION_RSP (60, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
MUTATION_REQ (0, P3, writeTimeout, MUTATION, () -> Mutation.serializer, () -> MutationVerbHandler.instance, MUTATION_RSP ),
|
|
VIRTUAL_MUTATION_RSP (200, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
VIRTUAL_MUTATION_REQ (201, P3, writeTimeout, MUTATION, () -> VirtualMutation.serializer, () -> VirtualMutation.handler, VIRTUAL_MUTATION_RSP),
|
|
HINT_RSP (61, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
HINT_REQ (1, P4, writeTimeout, MUTATION, () -> HintMessage.serializer, () -> HintVerbHandler.instance, HINT_RSP ),
|
|
READ_REPAIR_RSP (62, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
READ_REPAIR_REQ (2, P1, writeTimeout, MUTATION, () -> Mutation.serializer, () -> ReadRepairVerbHandler.instance, READ_REPAIR_RSP ),
|
|
BATCH_STORE_RSP (65, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
BATCH_STORE_REQ (5, P3, writeTimeout, MUTATION, () -> Batch.serializer, () -> BatchStoreVerbHandler.instance, BATCH_STORE_RSP ),
|
|
BATCH_REMOVE_RSP (66, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
BATCH_REMOVE_REQ (6, P3, writeTimeout, MUTATION, () -> TimeUUID.Serializer.instance, () -> BatchRemoveVerbHandler.instance, BATCH_REMOVE_RSP ),
|
|
|
|
PAXOS_PREPARE_RSP (93, P2, writeTimeout, REQUEST_RESPONSE, () -> PrepareResponse.serializer, RESPONSE_HANDLER ),
|
|
PAXOS_PREPARE_REQ (33, P2, writeTimeout, MUTATION, () -> Commit.serializer, () -> PrepareVerbHandler.instance, PAXOS_PREPARE_RSP ),
|
|
PAXOS_PROPOSE_RSP (94, P2, writeTimeout, REQUEST_RESPONSE, () -> BooleanSerializer.messagingSerializer, RESPONSE_HANDLER ),
|
|
PAXOS_PROPOSE_REQ (34, P2, writeTimeout, MUTATION, () -> Commit.serializer, () -> ProposeVerbHandler.instance, PAXOS_PROPOSE_RSP ),
|
|
PAXOS_COMMIT_RSP (95, P2, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS_COMMIT_REQ (35, P2, writeTimeout, MUTATION, () -> Agreed.serializer, () -> PaxosCommit.requestHandler, PAXOS_COMMIT_RSP ),
|
|
PAXOS_COMMIT_FORWARD_RSP (96, P2, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS_COMMIT_FORWARD_REQ (32, P2, writeTimeout, TRACKED_CAS, () -> PaxosCommitForwardRequest.serializer, () -> PaxosCommitForwardHandler.instance, PAXOS_COMMIT_FORWARD_RSP ),
|
|
|
|
TRUNCATE_RSP (79, P0, truncateTimeout, REQUEST_RESPONSE, () -> TruncateResponse.serializer, RESPONSE_HANDLER ),
|
|
TRUNCATE_REQ (19, P0, truncateTimeout, MUTATION, () -> TruncateRequest.serializer, () -> TruncateVerbHandler.instance, TRUNCATE_RSP ),
|
|
|
|
COUNTER_MUTATION_RSP (84, P1, counterTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
COUNTER_MUTATION_REQ (24, P2, counterTimeout, COUNTER_MUTATION, () -> CounterMutation.serializer, () -> CounterMutationVerbHandler.instance, COUNTER_MUTATION_RSP),
|
|
|
|
READ_RSP (63, P2, readTimeout, REQUEST_RESPONSE, () -> ReadResponse.serializer, RESPONSE_HANDLER ),
|
|
READ_REQ (3, P3, readTimeout, READ, () -> ReadCommand.serializer, () -> ReadCommandVerbHandler.instance, READ_RSP ),
|
|
RANGE_RSP (69, P2, rangeTimeout, REQUEST_RESPONSE, () -> ReadResponse.serializer, RESPONSE_HANDLER ),
|
|
RANGE_REQ (9, P3, rangeTimeout, READ, () -> ReadCommand.serializer, () -> ReadCommandVerbHandler.instance, RANGE_RSP ),
|
|
|
|
GOSSIP_DIGEST_SYN (14, P0, longTimeout, GOSSIP, () -> GossipDigestSyn.serializer, () -> GossipDigestSynVerbHandler.instance ),
|
|
GOSSIP_DIGEST_ACK (15, P0, longTimeout, GOSSIP, () -> GossipDigestAck.serializer, () -> GossipDigestAckVerbHandler.instance ),
|
|
GOSSIP_DIGEST_ACK2 (16, P0, longTimeout, GOSSIP, () -> GossipDigestAck2.serializer, () -> GossipDigestAck2VerbHandler.instance ),
|
|
GOSSIP_SHUTDOWN (29, P0, rpcTimeout, GOSSIP, () -> GossipShutdown.serializer, () -> GossipShutdownVerbHandler.instance ),
|
|
|
|
ECHO_RSP (91, P0, rpcTimeout, GOSSIP, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
ECHO_REQ (31, P0, rpcTimeout, GOSSIP, () -> NoPayload.serializer, () -> EchoVerbHandler.instance, ECHO_RSP ),
|
|
PING_RSP (97, P1, pingTimeout, GOSSIP, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PING_REQ (37, P1, pingTimeout, GOSSIP, () -> PingRequest.serializer, () -> PingVerbHandler.instance, PING_RSP ),
|
|
|
|
// P1 because messages can be arbitrarily large or aren't crucial
|
|
@Deprecated (since = "CEP-21")
|
|
SCHEMA_PUSH_RSP (98, P1, rpcTimeout, MIGRATION, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
@Deprecated (since = "CEP-21")
|
|
SCHEMA_PUSH_REQ (18, P1, rpcTimeout, MIGRATION, () -> SchemaMutationsSerializer.instance, () -> SchemaPushVerbHandler.instance, SCHEMA_PUSH_RSP ),
|
|
@Deprecated (since = "CEP-21")
|
|
SCHEMA_PULL_RSP (88, P1, rpcTimeout, MIGRATION, () -> SchemaMutationsSerializer.instance, RESPONSE_HANDLER ),
|
|
@Deprecated (since = "CEP-21")
|
|
SCHEMA_PULL_REQ (28, P1, rpcTimeout, MIGRATION, () -> NoPayload.serializer, () -> SchemaPullVerbHandler.instance, SCHEMA_PULL_RSP ),
|
|
SCHEMA_VERSION_RSP (80, P1, rpcTimeout, MIGRATION, () -> UUIDSerializer.serializer, RESPONSE_HANDLER ),
|
|
SCHEMA_VERSION_REQ (20, P1, rpcTimeout, MIGRATION, () -> NoPayload.serializer, () -> SchemaVersionVerbHandler.instance, SCHEMA_VERSION_RSP ),
|
|
|
|
// repair; mostly doesn't use callbacks and sends responses as their own request messages, with matching sessions by uuid; should eventually harmonize and make idiomatic
|
|
// for the repair messages that implement retry logic, use rpcTimeout so the single request fails faster, then retries can be used to recover
|
|
REPAIR_RSP (100, P1, repairTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
VALIDATION_RSP (102, P1, repairValidationRspTimeout, ANTI_ENTROPY, () -> ValidationResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
VALIDATION_REQ (101, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> ValidationRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
SYNC_RSP (104, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> SyncResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
SYNC_REQ (103, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> SyncRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
PREPARE_MSG (105, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
SNAPSHOT_MSG (106, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> SnapshotMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
CLEANUP_MSG (107, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> CleanupMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
PREPARE_CONSISTENT_RSP (109, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareConsistentResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
PREPARE_CONSISTENT_REQ (108, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareConsistentRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
FINALIZE_PROPOSE_MSG (110, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizePropose.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
FINALIZE_PROMISE_MSG (111, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizePromise.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
FINALIZE_COMMIT_MSG (112, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizeCommit.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
FAILED_SESSION_MSG (113, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FailSession.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
STATUS_RSP (115, P1, repairTimeout, ANTI_ENTROPY, () -> StatusResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
STATUS_REQ (114, P1, repairTimeout, ANTI_ENTROPY, () -> StatusRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ),
|
|
MT_SYNC_RSP (117, P1, repairWithBackoffTimeout, REQUEST_RESPONSE, () -> MutationTrackingSyncResponse.serializer, RESPONSE_HANDLER ),
|
|
MT_SYNC_REQ (116, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> MutationTrackingSyncRequest.serializer, () -> RepairMessageVerbHandler.instance(), MT_SYNC_RSP ),
|
|
|
|
REPLICATION_DONE_RSP (82, P0, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
REPLICATION_DONE_REQ (22, P0, rpcTimeout, MISC, () -> NoPayload.serializer, () -> ReplicationDoneVerbHandler.instance, REPLICATION_DONE_RSP),
|
|
SNAPSHOT_RSP (87, P0, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
SNAPSHOT_REQ (27, P0, rpcTimeout, MISC, () -> SnapshotCommand.serializer, () -> SnapshotVerbHandler.instance, SNAPSHOT_RSP ),
|
|
|
|
PAXOS2_COMMIT_REMOTE_REQ (38, P2, writeTimeout, MUTATION, () -> Mutation.serializer, () -> MutationVerbHandler.instance, MUTATION_RSP ),
|
|
PAXOS2_COMMIT_REMOTE_RSP (39, P2, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_PREPARE_RSP (50, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepare.responseSerializer, RESPONSE_HANDLER ),
|
|
PAXOS2_PREPARE_REQ (40, P2, writeTimeout, MUTATION, () -> PaxosPrepare.requestSerializer, () -> PaxosPrepare.requestHandler, PAXOS2_PREPARE_RSP ),
|
|
PAXOS2_PREPARE_REFRESH_RSP (51, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepareRefresh.responseSerializer, RESPONSE_HANDLER ),
|
|
PAXOS2_PREPARE_REFRESH_REQ (41, P2, writeTimeout, MUTATION, () -> PaxosPrepareRefresh.requestSerializer, () -> PaxosPrepareRefresh.requestHandler, PAXOS2_PREPARE_REFRESH_RSP ),
|
|
PAXOS2_PROPOSE_RSP (52, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPropose.ACCEPT_RESULT_SERIALIZER, RESPONSE_HANDLER ),
|
|
PAXOS2_PROPOSE_REQ (42, P2, writeTimeout, MUTATION, () -> PaxosPropose.requestSerializer, () -> PaxosPropose.requestHandler, PAXOS2_PROPOSE_RSP ),
|
|
PAXOS2_COMMIT_AND_PREPARE_RSP (53, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepare.responseSerializer, RESPONSE_HANDLER ),
|
|
PAXOS2_COMMIT_AND_PREPARE_REQ (43, P2, writeTimeout, MUTATION, () -> PaxosCommitAndPrepare.requestSerializer, () -> PaxosCommitAndPrepare.requestHandler, PAXOS2_COMMIT_AND_PREPARE_RSP ),
|
|
PAXOS2_REPAIR_RSP (54, P2, writeTimeout, PAXOS_REPAIR, () -> PaxosRepair.responseSerializer, RESPONSE_HANDLER ),
|
|
PAXOS2_REPAIR_REQ (44, P2, writeTimeout, PAXOS_REPAIR, () -> PaxosRepair.requestSerializer, () -> PaxosRepair.requestHandler, PAXOS2_REPAIR_RSP ),
|
|
PAXOS2_CLEANUP_START_PREPARE_RSP (55, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosCleanupHistory.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_CLEANUP_START_PREPARE_REQ (45, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosStartPrepareCleanup.serializer, () -> PaxosStartPrepareCleanup.verbHandler, PAXOS2_CLEANUP_START_PREPARE_RSP ),
|
|
PAXOS2_CLEANUP_RSP (56, P2, repairTimeout, PAXOS_REPAIR, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_CLEANUP_REQ (46, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosCleanupRequest.serializer, () -> PaxosCleanupRequest.verbHandler, PAXOS2_CLEANUP_RSP ),
|
|
PAXOS2_CLEANUP_RSP2 (57, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosCleanupResponse.serializer, () -> PaxosCleanupResponse.verbHandler ),
|
|
PAXOS2_CLEANUP_FINISH_PREPARE_RSP(58, P2, repairTimeout, PAXOS_REPAIR, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_CLEANUP_FINISH_PREPARE_REQ(47, P2, repairTimeout, IMMEDIATE, () -> PaxosCleanupHistory.serializer, () -> PaxosFinishPrepareCleanup.verbHandler, PAXOS2_CLEANUP_FINISH_PREPARE_RSP),
|
|
PAXOS2_CLEANUP_COMPLETE_RSP (59, P2, repairTimeout, PAXOS_REPAIR, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_CLEANUP_COMPLETE_REQ (48, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosCleanupComplete.serializer, () -> PaxosCleanupComplete.verbHandler, PAXOS2_CLEANUP_COMPLETE_RSP ),
|
|
PAXOS2_UPDATE_LOW_BALLOT_RSP (67, P2, repairTimeout, PAXOS_REPAIR, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_UPDATE_LOW_BALLOT_REQ (64, P2, repairTimeout, PAXOS_REPAIR, () -> PaxosUpdateLowBallot.serializer, () -> PaxosUpdateLowBallot.verbHandler, PAXOS2_UPDATE_LOW_BALLOT_RSP ),
|
|
PAXOS2_COMMIT_FORWARD_RSP (71, P2, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
PAXOS2_COMMIT_FORWARD_REQ (72, P2, writeTimeout, TRACKED_CAS, () -> Paxos2CommitForwardRequest.serializer, () -> Paxos2CommitForwardHandler.instance, PAXOS2_COMMIT_FORWARD_RSP ),
|
|
|
|
// CAS and consensus read forwarding for tracked keyspaces
|
|
CAS_FORWARD_RSP (73, P2, writeTimeout, REQUEST_RESPONSE, () -> CasForwardResponse.serializer, RESPONSE_HANDLER ),
|
|
CAS_FORWARD_REQ (74, P2, writeTimeout, TRACKED_CAS, () -> CasForwardRequest.serializer, () -> CasForwardHandler.instance, CAS_FORWARD_RSP ),
|
|
CONSENSUS_READ_FORWARD_RSP (75, P2, readTimeout, REQUEST_RESPONSE, () -> CasForwardResponse.serializer, RESPONSE_HANDLER ),
|
|
CONSENSUS_READ_FORWARD_REQ (76, P2, readTimeout, TRACKED_CAS, () -> ConsensusReadForwardRequest.serializer, () -> ConsensusReadForwardHandler.instance, CONSENSUS_READ_FORWARD_RSP ),
|
|
PAXOS_PREPARE_REFRESH_FORWARD_RSP(77, P2, writeTimeout, REQUEST_RESPONSE, () -> PrepareRefreshForwardResponse.serializer,RESPONSE_HANDLER ),
|
|
PAXOS_PREPARE_REFRESH_FORWARD_REQ(78, P2, writeTimeout, TRACKED_CAS, () -> PrepareRefreshForwardRequest.serializer, () -> PrepareRefreshForwardHandler.instance, PAXOS_PREPARE_REFRESH_FORWARD_RSP),
|
|
|
|
// transactional cluster metadata
|
|
TCM_COMMIT_RSP (801, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::commitResultSerializer, RESPONSE_HANDLER ),
|
|
TCM_COMMIT_REQ (802, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::commitSerializer, () -> commitRequestHandler(), TCM_COMMIT_RSP ),
|
|
TCM_FETCH_CMS_LOG_RSP (803, P0, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
|
|
TCM_FETCH_CMS_LOG_REQ (804, P0, rpcTimeout, FETCH_METADATA, () -> FetchCMSLog.serializer, () -> fetchLogRequestHandler(), TCM_FETCH_CMS_LOG_RSP ),
|
|
TCM_REPLICATION (805, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::logStateSerializer, () -> replicationHandler() ),
|
|
TCM_NOTIFY_RSP (806, P0, rpcTimeout, INTERNAL_METADATA, () -> Epoch.messageSerializer, RESPONSE_HANDLER ),
|
|
TCM_NOTIFY_REQ (807, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::logStateSerializer, () -> logNotifyHandler(), TCM_NOTIFY_RSP ),
|
|
TCM_CURRENT_EPOCH_REQ (808, P0, rpcTimeout, INTERNAL_METADATA, () -> Epoch.messageSerializer, () -> currentEpochRequestHandler(), TCM_NOTIFY_RSP ),
|
|
TCM_INIT_MIG_RSP (809, P0, rpcTimeout, INTERNAL_METADATA, () -> CMSInitializationResponse.serializer, RESPONSE_HANDLER ),
|
|
TCM_INIT_MIG_REQ (810, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::initRequestSerializer, () -> Election.instance.prepareHandler, TCM_INIT_MIG_RSP ),
|
|
TCM_ABORT_MIG (811, P0, rpcTimeout, INTERNAL_METADATA, () -> CMSInitializationRequest.Initiator.serializer,() -> Election.instance.abortHandler, TCM_INIT_MIG_RSP ),
|
|
TCM_DISCOVER_RSP (812, P0, rpcTimeout, INTERNAL_METADATA, () -> Discovery.serializer, RESPONSE_HANDLER ),
|
|
TCM_DISCOVER_REQ (813, P0, rpcTimeout, INTERNAL_METADATA, () -> NoPayload.serializer, () -> Discovery.instance.requestHandler, TCM_DISCOVER_RSP ),
|
|
TCM_FETCH_PEER_LOG_RSP (818, P0, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
|
|
TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout, FETCH_METADATA, () -> FetchPeerLog.serializer, () -> FetchPeerLog.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),
|
|
|
|
INITIATE_DATA_MOVEMENTS_RSP (814, P1, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
INITIATE_DATA_MOVEMENTS_REQ (815, P1, rpcTimeout, MISC, () -> DataMovement.serializer, () -> DataMovementVerbHandler.instance, INITIATE_DATA_MOVEMENTS_RSP ),
|
|
DATA_MOVEMENT_EXECUTED_RSP (816, P1, rpcTimeout, MISC, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
DATA_MOVEMENT_EXECUTED_REQ (817, P1, rpcTimeout, MISC, () -> DataMovement.Status.serializer, () -> DataMovements.instance, DATA_MOVEMENT_EXECUTED_RSP ),
|
|
|
|
// tracked replication
|
|
PULL_MUTATIONS_REQ (901, P2, readTimeout, READ, () -> PullMutationsRequest.serializer, () -> PullMutationsRequest.verbHandler ),
|
|
PUSH_MUTATION_REQ (902, P3, writeTimeout, MUTATION, () -> PushMutationRequest.serializer, () -> PushMutationRequest.verbHandler, MUTATION_RSP ),
|
|
READ_RECONCILE_ACK (903, P2, readTimeout, REQUEST_RESPONSE, () -> ReadReconcileAck.serializer, () -> ReadReconcileAck.verbHandler ),
|
|
FORWARD_WRITE_REQ (904, P3, writeTimeout, MUTATION, () -> ForwardedWrite.serializer, () -> ForwardedWrite.verbHandler ),
|
|
BROADCAST_LOG_OFFSETS (905, P1, rpcTimeout, MISC, () -> BroadcastLogOffsets.serializer, () -> BroadcastLogOffsets.verbHandler ),
|
|
|
|
TRACKED_PARTITION_READ_RSP (906, P2, readTimeout, REQUEST_RESPONSE, () -> TrackedDataResponse.serializer, RESPONSE_HANDLER ),
|
|
TRACKED_PARTITION_READ_REQ (907, P3, readTimeout, READ, () -> TrackedRead.DataRequest.serializer, () -> TrackedRead.verbHandler, TRACKED_PARTITION_READ_RSP),
|
|
TRACKED_RANGE_READ_RSP (908, P2, rangeTimeout, REQUEST_RESPONSE, () -> TrackedDataResponse.serializer, RESPONSE_HANDLER ),
|
|
TRACKED_RANGE_READ_REQ (909, P3, rangeTimeout, READ, () -> TrackedRead.DataRequest.serializer, () -> TrackedRead.verbHandler, TRACKED_RANGE_READ_RSP ),
|
|
TRACKED_SUMMARY_RSP (910, P2, readTimeout, REQUEST_RESPONSE, () -> TrackedSummaryResponse.serializer, () -> TrackedSummaryResponse.verbHandler ),
|
|
TRACKED_SUMMARY_REQ (911, P3, readTimeout, READ, () -> TrackedRead.SummaryRequest.serializer, () -> TrackedRead.verbHandler, TRACKED_SUMMARY_RSP ),
|
|
|
|
TRACKED_TRANSFER_ACTIVATE_RSP (912, P1, repairTimeout, REQUEST_RESPONSE, () -> ActivationResponse.serializer, RESPONSE_HANDLER),
|
|
TRACKED_TRANSFER_ACTIVATE_REQ (913, P1, repairTimeout, ANTI_ENTROPY, () -> ActivationRequest.serializer, () -> ActivationRequest.verbHandler, TRACKED_TRANSFER_ACTIVATE_RSP),
|
|
TRACKED_TRANSFER_FAILED_RSP (914, P1, repairTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER),
|
|
TRACKED_TRANSFER_FAILED_REQ (915, P1, repairTimeout, ANTI_ENTROPY, () -> TransferFailed.serializer, () -> TransferTrackingService.verbHandler, TRACKED_TRANSFER_FAILED_RSP),
|
|
|
|
// accord
|
|
ACCORD_SIMPLE_RSP (119, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(EnumSerializer.simpleReply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_PRE_ACCEPT_RSP (120, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(PreacceptSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_PRE_ACCEPT_REQ (121, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(PreacceptSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_PRE_ACCEPT_RSP ),
|
|
ACCORD_ACCEPT_RSP (122, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AcceptSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_ACCEPT_REQ (123, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AcceptSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ),
|
|
ACCORD_NOT_ACCEPT_REQ (124, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AcceptSerializers.notAccept), AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ),
|
|
ACCORD_READ_RSP (125, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializer.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_READ_REQ (126, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializer.request), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
|
|
ACCORD_STABLE_THEN_READ_REQ (127, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializer.request), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
|
|
ACCORD_COMMIT_REQ (128, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CommitSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
|
|
ACCORD_COMMIT_INVALIDATE_REQ (129, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CommitSerializers.invalidate), AccordService::requestHandlerOrNoop ),
|
|
ACCORD_APPLY_RSP (130, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ApplySerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_APPLY_REQ (131, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ApplySerializers.request), AccordService::requestHandlerOrNoop, ACCORD_APPLY_RSP ),
|
|
ACCORD_APPLY_AND_WAIT_REQ (132, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializer.request), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP),
|
|
ACCORD_BEGIN_RECOVER_RSP (133, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(RecoverySerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_BEGIN_RECOVER_REQ (134, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(RecoverySerializers.request), AccordService::requestHandlerOrNoop, ACCORD_BEGIN_RECOVER_RSP ),
|
|
ACCORD_BEGIN_INVALIDATE_RSP (135, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(BeginInvalidationSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_BEGIN_INVALIDATE_REQ (136, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(BeginInvalidationSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_BEGIN_INVALIDATE_RSP ),
|
|
ACCORD_AWAIT_RSP (137, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.syncReply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_AWAIT_REQ (138, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_AWAIT_RSP ),
|
|
ACCORD_AWAIT_ASYNC_RSP_REQ (139, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.asyncReply), AccordService::requestHandlerOrNoop ),
|
|
ACCORD_WAIT_UNTIL_APPLIED_REQ (140, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializer.request), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
|
|
ACCORD_RECOVER_AWAIT_RSP (141, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverReply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_RECOVER_AWAIT_REQ (142, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverRequest), AccordService::requestHandlerOrNoop, ACCORD_RECOVER_AWAIT_RSP),
|
|
ACCORD_INFORM_DURABLE_REQ (143, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(InformSerializers.durable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
|
|
ACCORD_CHECK_STATUS_RSP (144, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_CHECK_STATUS_REQ (145, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ),
|
|
ACCORD_FETCH_DATA_RSP (146, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(FetchSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_FETCH_DATA_REQ (147, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(FetchSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_FETCH_DATA_RSP ),
|
|
ACCORD_GET_EPHMRL_READ_DEPS_RSP (148, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetEphmrlReadDepsSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_GET_EPHMRL_READ_DEPS_REQ (149, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetEphmrlReadDepsSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_EPHMRL_READ_DEPS_RSP),
|
|
ACCORD_GET_LATEST_DEPS_RSP (150, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(LatestDepsSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_GET_LATEST_DEPS_REQ (151, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(LatestDepsSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_LATEST_DEPS_RSP),
|
|
ACCORD_GET_MAX_CONFLICT_RSP (152, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_GET_MAX_CONFLICT_REQ (153, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP),
|
|
ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ),
|
|
ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ),
|
|
ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
|
|
ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
|
|
|
|
ACCORD_SYNC_NOTIFY_RSP (158, P2, writeTimeout, MISC, () -> accordEmbedded(EnumSerializer.simpleReply), RESPONSE_HANDLER),
|
|
ACCORD_SYNC_NOTIFY_REQ (159, P2, writeTimeout, MISC, () -> accordEmbedded(Notification.serializer), () -> AccordSyncPropagator.verbHandler, ACCORD_SYNC_NOTIFY_RSP ),
|
|
|
|
CONSENSUS_KEY_MIGRATION (160, P1, writeTimeout, MISC, () -> accordEmbedded(ConsensusKeyMigrationFinished.serializer),() -> ConsensusKeyMigrationState.consensusKeyMigrationFinishedHandler),
|
|
|
|
ACCORD_INTEROP_READ_RSP (161, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropRead.replySerializer), AccordService::responseHandlerOrNoop),
|
|
ACCORD_INTEROP_READ_REQ (162, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropRead.requestSerializer), AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP),
|
|
ACCORD_INTEROP_STABLE_THEN_READ_REQ(163, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropStableThenRead.requestSerializer), AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP),
|
|
ACCORD_INTEROP_READ_REPAIR_RSP (164, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropReadRepair.replySerializer), AccordService::responseHandlerOrNoop),
|
|
ACCORD_INTEROP_READ_REPAIR_REQ (165, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropReadRepair.requestSerializer), AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_REPAIR_RSP),
|
|
ACCORD_INTEROP_APPLY_REQ (166, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropApply.serializer), AccordService::requestHandlerOrNoop, ACCORD_APPLY_RSP),
|
|
ACCORD_FETCH_WATERMARKS_RSP (167, P0, shortTimeout, FETCH_METADATA, () -> accordEmbedded(WatermarkCollector.serializer), RESPONSE_HANDLER),
|
|
// NoPayload can not be prefixed with accord version as it is special cased in C* messaging
|
|
ACCORD_FETCH_WATERMARKS_REQ (168, P0, shortTimeout, FETCH_METADATA, () -> NoPayload.serializer, AccordService::watermarkHandlerOrNoop, ACCORD_FETCH_WATERMARKS_RSP),
|
|
ACCORD_FETCH_TOPOLOGY_RSP (169, P0, shortTimeout, FETCH_METADATA, () -> accordEmbedded(FetchTopologies.responseSerializer), RESPONSE_HANDLER),
|
|
ACCORD_FETCH_TOPOLOGY_REQ (170, P0, shortTimeout, FETCH_METADATA, () -> accordEmbedded(FetchTopologies.serializer), () -> FetchTopologies.handler, ACCORD_FETCH_TOPOLOGY_RSP),
|
|
|
|
DICTIONARY_UPDATE_RSP (171, P1, rpcTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
|
|
DICTIONARY_UPDATE_REQ (172, P1, rpcTimeout, MISC, () -> CompressionDictionaryUpdateMessage.serializer, () -> CompressionDictionaryUpdateVerbHandler.instance, DICTIONARY_UPDATE_RSP ),
|
|
|
|
// generic failure response
|
|
FAILURE_RSP (99, P0, noTimeout, REQUEST_RESPONSE, () -> RequestFailure.serializer, RESPONSE_HANDLER ),
|
|
|
|
// dummy verbs
|
|
_TRACE (30, P1, rpcTimeout, TRACING, () -> NoPayload.serializer, () -> null ),
|
|
_SAMPLE (49, P1, rpcTimeout, INTERNAL_RESPONSE, () -> NoPayload.serializer, () -> null ),
|
|
_TEST_1 (10, P0, writeTimeout, IMMEDIATE, () -> NoPayload.serializer, () -> null ),
|
|
_TEST_2 (11, P1, rpcTimeout, IMMEDIATE, () -> NoPayload.serializer, () -> null ),
|
|
|
|
/** @deprecated See CASSANDRA-15066 */
|
|
@Deprecated(since = "4.0")
|
|
REQUEST_RSP (4, P1, rpcTimeout, REQUEST_RESPONSE, () -> null, () -> ResponseVerbHandler.instance ),
|
|
/** @deprecated See CASSANDRA-15066 */
|
|
@Deprecated(since = "4.0")
|
|
INTERNAL_RSP (23, P1, rpcTimeout, INTERNAL_RESPONSE, () -> null, () -> ResponseVerbHandler.instance ),
|
|
|
|
// CUSTOM VERBS
|
|
UNUSED_CUSTOM_VERB (CUSTOM,
|
|
0, P1, rpcTimeout, INTERNAL_RESPONSE, () -> null, () -> null ),
|
|
;
|
|
|
|
public static final List<Verb> VERBS = ImmutableList.copyOf(Verb.values());
|
|
|
|
public enum Priority
|
|
{
|
|
P0, // sends on the urgent connection (i.e. for Gossip, Echo)
|
|
P1, // small or empty responses
|
|
P2, // larger messages that can be dropped but who have a larger impact on system stability (e.g. READ_REPAIR, READ_RSP)
|
|
P3,
|
|
P4
|
|
}
|
|
|
|
public enum Kind
|
|
{
|
|
NORMAL,
|
|
CUSTOM
|
|
}
|
|
|
|
public final int id;
|
|
public final Priority priority;
|
|
public final Stage stage;
|
|
public final Kind kind;
|
|
|
|
/**
|
|
* Messages we receive from peers have a Verb that tells us what kind of message it is.
|
|
* Most of the time, this is enough to determine how to deserialize the message payload.
|
|
* The exception is the REQUEST_RSP verb, which just means "a response to something you told me to do."
|
|
* Traditionally, this was fine since each VerbHandler knew what type of payload it expected, and
|
|
* handled the deserialization itself. Now that we do that in ITC, to avoid the extra copy to an
|
|
* intermediary byte[] (See CASSANDRA-3716), we need to wire that up to the CallbackInfo object
|
|
* (see below).
|
|
*
|
|
* NOTE: we use a Supplier to avoid loading the dependent classes until necessary.
|
|
*/
|
|
private final Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer;
|
|
private final Supplier<? extends IVerbHandler<?>> handler;
|
|
|
|
public final Verb responseVerb;
|
|
|
|
private final ToLongFunction<TimeUnit> expiration;
|
|
|
|
|
|
/**
|
|
* Verbs it's okay to drop if the request has been queued longer than the request timeout. These
|
|
* all correspond to client requests or something triggered by them; we don't want to
|
|
* drop internal messages like bootstrap or repair notifications.
|
|
*/
|
|
Verb(int id, Priority priority, ToLongFunction<TimeUnit> expiration, Stage stage, Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer, Supplier<? extends IVerbHandler<?>> handler)
|
|
{
|
|
this(NORMAL, id, priority, expiration, stage, serializer, handler, null);
|
|
}
|
|
|
|
Verb(int id, Priority priority, ToLongFunction<TimeUnit> expiration, Stage stage, Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer, Supplier<? extends IVerbHandler<?>> handler, Verb responseVerb)
|
|
{
|
|
this(NORMAL, id, priority, expiration, stage, serializer, handler, responseVerb);
|
|
}
|
|
|
|
Verb(Kind kind, int id, Priority priority, ToLongFunction<TimeUnit> expiration, Stage stage, Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer, Supplier<? extends IVerbHandler<?>> handler)
|
|
{
|
|
this(kind, id, priority, expiration, stage, serializer, handler, null);
|
|
}
|
|
|
|
Verb(Kind kind, int id, Priority priority, ToLongFunction<TimeUnit> expiration, Stage stage, Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer, Supplier<? extends IVerbHandler<?>> handler, Verb responseVerb)
|
|
{
|
|
this.stage = stage;
|
|
if (id < 0)
|
|
throw new IllegalArgumentException("Verb id must be non-negative, got " + id + " for verb " + name());
|
|
|
|
if (kind == CUSTOM)
|
|
{
|
|
if (id > MAX_CUSTOM_VERB_ID)
|
|
throw new AssertionError("Invalid custom verb id " + id + " - we only allow custom ids between 0 and " + MAX_CUSTOM_VERB_ID);
|
|
this.id = idForCustomVerb(id);
|
|
}
|
|
else
|
|
{
|
|
if (id > CUSTOM_VERB_START - MAX_CUSTOM_VERB_ID)
|
|
throw new AssertionError("Invalid verb id " + id + " - we only allow ids between 0 and " + (CUSTOM_VERB_START - MAX_CUSTOM_VERB_ID));
|
|
this.id = id;
|
|
}
|
|
this.priority = priority;
|
|
this.serializer = serializer;
|
|
this.handler = handler;
|
|
this.responseVerb = responseVerb;
|
|
this.expiration = expiration;
|
|
this.kind = kind;
|
|
}
|
|
|
|
public <In, Out> IVersionedAsymmetricSerializer<In, Out> serializer()
|
|
{
|
|
return (IVersionedAsymmetricSerializer<In, Out>) serializer.get();
|
|
}
|
|
|
|
public <T> IVerbHandler<T> handler()
|
|
{
|
|
return (IVerbHandler<T>) handler.get();
|
|
}
|
|
|
|
public long expiresAtNanos(long nowNanos)
|
|
{
|
|
return nowNanos + expiresAfterNanos();
|
|
}
|
|
|
|
public long expiresAfterNanos()
|
|
{
|
|
return expiration.applyAsLong(NANOSECONDS);
|
|
}
|
|
|
|
public long expiresAfter(TimeUnit units)
|
|
{
|
|
return expiration.applyAsLong(units);
|
|
}
|
|
|
|
// this is a little hacky, but reduces the number of parameters up top
|
|
public boolean isResponse()
|
|
{
|
|
return handler.get() == ResponseVerbHandler.instance;
|
|
}
|
|
|
|
@VisibleForTesting
|
|
Supplier<? extends IVerbHandler<?>> unsafeSetHandler(Supplier<? extends IVerbHandler<?>> handler) throws NoSuchFieldException, IllegalAccessException
|
|
{
|
|
Supplier<? extends IVerbHandler<?>> original = this.handler;
|
|
Field field = Verb.class.getDeclaredField("handler");
|
|
field.setAccessible(true);
|
|
Field modifiers = ReflectionUtils.getModifiersField();
|
|
modifiers.setAccessible(true);
|
|
modifiers.setInt(field, field.getModifiers() & ~Modifier.FINAL);
|
|
field.set(this, handler);
|
|
return original;
|
|
}
|
|
|
|
@VisibleForTesting
|
|
public Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> unsafeSetSerializer(Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> serializer) throws NoSuchFieldException, IllegalAccessException
|
|
{
|
|
Supplier<? extends IVersionedAsymmetricSerializer<?, ?>> original = this.serializer;
|
|
Field field = Verb.class.getDeclaredField("serializer");
|
|
field.setAccessible(true);
|
|
Field modifiers = ReflectionUtils.getModifiersField();
|
|
modifiers.setAccessible(true);
|
|
modifiers.setInt(field, field.getModifiers() & ~Modifier.FINAL);
|
|
field.set(this, serializer);
|
|
return original;
|
|
}
|
|
|
|
@VisibleForTesting
|
|
ToLongFunction<TimeUnit> unsafeSetExpiration(ToLongFunction<TimeUnit> expiration) throws NoSuchFieldException, IllegalAccessException
|
|
{
|
|
ToLongFunction<TimeUnit> original = this.expiration;
|
|
Field field = Verb.class.getDeclaredField("expiration");
|
|
field.setAccessible(true);
|
|
Field modifiers = ReflectionUtils.getModifiersField();
|
|
modifiers.setAccessible(true);
|
|
modifiers.setInt(field, field.getModifiers() & ~Modifier.FINAL);
|
|
field.set(this, expiration);
|
|
return original;
|
|
}
|
|
|
|
// This is the largest number we can store in 2 bytes using VIntCoding (1 bit per byte is used to indicate if there is more data coming).
|
|
// When generating ids we count *down* from this number
|
|
private static final int CUSTOM_VERB_START = (1 << (7 * 2)) - 1;
|
|
|
|
// Sanity check for the custom verb ids - avoids someone mistakenly adding a custom verb id close to the normal verbs which
|
|
// could cause a conflict later when new normal verbs are added.
|
|
private static final int MAX_CUSTOM_VERB_ID = 1000;
|
|
|
|
private static final Verb[] idToVerbMap;
|
|
private static final Verb[] idToCustomVerbMap;
|
|
private static final int minCustomId;
|
|
|
|
static
|
|
{
|
|
Verb[] verbs = values();
|
|
int max = -1;
|
|
int minCustom = Integer.MAX_VALUE;
|
|
for (Verb v : verbs)
|
|
{
|
|
switch (v.kind)
|
|
{
|
|
case NORMAL:
|
|
max = Math.max(v.id, max);
|
|
break;
|
|
case CUSTOM:
|
|
minCustom = Math.min(v.id, minCustom);
|
|
break;
|
|
default:
|
|
throw new AssertionError("Unsupported Verb Kind: " + v.kind + " for verb " + v);
|
|
}
|
|
}
|
|
minCustomId = minCustom;
|
|
|
|
if (minCustom <= max)
|
|
throw new IllegalStateException("Overlapping verb ids are not allowed");
|
|
|
|
Verb[] idMap = new Verb[max + 1];
|
|
int customCount = minCustom < Integer.MAX_VALUE ? CUSTOM_VERB_START - minCustom : 0;
|
|
Verb[] customIdMap = new Verb[customCount + 1];
|
|
for (Verb v : verbs)
|
|
{
|
|
switch (v.kind)
|
|
{
|
|
case NORMAL:
|
|
if (idMap[v.id] != null)
|
|
throw new IllegalArgumentException("cannot have two verbs that map to the same id: " + v + " and " + idMap[v.id]);
|
|
idMap[v.id] = v;
|
|
break;
|
|
case CUSTOM:
|
|
int relativeId = idForCustomVerb(v.id);
|
|
if (customIdMap[relativeId] != null)
|
|
throw new IllegalArgumentException("cannot have two custom verbs that map to the same id: " + v + " and " + customIdMap[relativeId]);
|
|
customIdMap[relativeId] = v;
|
|
break;
|
|
default:
|
|
throw new AssertionError("Unsupported Verb Kind: " + v.kind + " for verb " + v);
|
|
}
|
|
}
|
|
|
|
idToVerbMap = idMap;
|
|
idToCustomVerbMap = customIdMap;
|
|
}
|
|
|
|
public static Verb fromId(int id)
|
|
{
|
|
Verb[] verbs = idToVerbMap;
|
|
if (id >= minCustomId)
|
|
{
|
|
id = idForCustomVerb(id);
|
|
verbs = idToCustomVerbMap;
|
|
}
|
|
Verb verb = id >= 0 && id < verbs.length ? verbs[id] : null;
|
|
if (verb == null)
|
|
throw new IllegalArgumentException("Unknown verb id " + id);
|
|
return verb;
|
|
}
|
|
|
|
/**
|
|
* calculate an id for a custom verb
|
|
*/
|
|
private static int idForCustomVerb(int id)
|
|
{
|
|
return CUSTOM_VERB_START - id;
|
|
}
|
|
|
|
private static <A, B> IVersionedAsymmetricSerializer<A, B> accordEmbedded(AsymmetricVersionedSerializer<A, B, Version> delegate)
|
|
{
|
|
return AccordSerializers.embedded(Version.CLUSTER_SAFE_VERSION, delegate);
|
|
}
|
|
|
|
private static <A, B> IVersionedAsymmetricSerializer<A, B> accordEmbedded(AsymmetricUnversionedSerializer<A, B> delegate)
|
|
{
|
|
return accordEmbedded(AsymmetricVersionedSerializer.from(delegate));
|
|
}
|
|
}
|
|
|
|
@SuppressWarnings("unused")
|
|
class VerbTimeouts
|
|
{
|
|
static final ToLongFunction<TimeUnit> rpcTimeout = DatabaseDescriptor::getRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> shortTimeout = DatabaseDescriptor::getShortRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> writeTimeout = DatabaseDescriptor::getWriteRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> readTimeout = DatabaseDescriptor::getReadRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> rangeTimeout = DatabaseDescriptor::getRangeRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> counterTimeout = DatabaseDescriptor::getCounterWriteRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> truncateTimeout = DatabaseDescriptor::getTruncateRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> repairTimeout = DatabaseDescriptor::getRepairRpcTimeout;
|
|
static final ToLongFunction<TimeUnit> pingTimeout = DatabaseDescriptor::getPingTimeout;
|
|
static final ToLongFunction<TimeUnit> longTimeout = units -> Math.max(DatabaseDescriptor.getRpcTimeout(units), units.convert(5L, TimeUnit.MINUTES));
|
|
static final ToLongFunction<TimeUnit> noTimeout = units -> { throw new IllegalStateException(); };
|
|
|
|
// repair verbs need to have different timeouts based off if retries are enabled or not
|
|
static final ToLongFunction<TimeUnit> repairWithBackoffTimeout = units -> {
|
|
if (!DatabaseDescriptor.getRepairRetrySpec().isEnabled())
|
|
return repairTimeout.applyAsLong(units);
|
|
return rpcTimeout.applyAsLong(units);
|
|
};
|
|
static final ToLongFunction<TimeUnit> repairValidationRspTimeout = units -> {
|
|
if (!DatabaseDescriptor.getRepairRetrySpec().isMerkleTreeRetriesEnabled())
|
|
return longTimeout.applyAsLong(units);
|
|
return rpcTimeout.applyAsLong(units);
|
|
};
|
|
}
|
|
|
|
class ResponseHandlerSupplier
|
|
{
|
|
static final Supplier<IVerbHandler<?>> RESPONSE_HANDLER = () -> ResponseVerbHandler.instance;
|
|
}
|