import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.eclipse.jdt.annotation.NonNull;
import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.controller.cluster.access.ABIVersion;
import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
-import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
import org.opendaylight.controller.cluster.access.concepts.RequestException;
import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
import org.opendaylight.yangtools.yang.data.tree.api.DataValidationFailedException;
import org.opendaylight.yangtools.yang.data.tree.api.TreeType;
import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
-import org.opendaylight.yangtools.yang.model.api.EffectiveModelContextProvider;
import scala.concurrent.duration.FiniteDuration;
/**
private DatastoreContext datastoreContext;
+ @Deprecated(since = "9.0.0", forRemoval = true)
private final ShardCommitCoordinator commitCoordinator;
private long transactionCommitTimeout;
private final MessageTracker appendEntriesReplyTracker;
+ @Deprecated(since = "9.0.0", forRemoval = true)
private final ShardTransactionActorFactory transactionActorFactory;
private final ShardSnapshotCohort snapshotCohort;
private ShardSnapshot restoreFromSnapshot;
+ @Deprecated(since = "9.0.0", forRemoval = true)
private final ShardTransactionMessageRetrySupport messageRetrySupport;
@VisibleForTesting
handleRequestAssemblerMessage(message);
} else if (message instanceof ConnectClientRequest request) {
handleConnectClient(request);
- } else if (CreateTransaction.isSerializedType(message)) {
- handleCreateTransaction(message);
- } else if (message instanceof BatchedModifications request) {
- handleBatchedModifications(request);
- } else if (message instanceof ForwardedReadyTransaction request) {
- handleForwardedReadyTransaction(request);
- } else if (message instanceof ReadyLocalTransaction request) {
- handleReadyLocalTransaction(request);
- } else if (CanCommitTransaction.isSerializedType(message)) {
- handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
- } else if (CommitTransaction.isSerializedType(message)) {
- handleCommitTransaction(CommitTransaction.fromSerializable(message));
- } else if (AbortTransaction.isSerializedType(message)) {
- handleAbortTransaction(AbortTransaction.fromSerializable(message));
- } else if (CloseTransactionChain.isSerializedType(message)) {
- closeTransactionChain(CloseTransactionChain.fromSerializable(message));
} else if (message instanceof DataTreeChangedReply) {
// Ignore reply
} else if (message instanceof RegisterDataTreeChangeListener request) {
sender().tell(store.getDataTree(), self());
} else if (message instanceof ServerRemoved) {
context().parent().forward(message, context());
- } else if (ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) {
- messageRetrySupport.onTimerMessage(message);
} else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand request) {
store.processCohortRegistryCommand(getSender(), request);
} else if (message instanceof MakeLeaderLocal) {
} else if (GetKnownClients.INSTANCE.equals(message)) {
handleGetKnownClients();
} else if (!responseMessageSlicer.handleMessage(message)) {
- super.handleNonRaftCommand(message);
+ // Ask-based protocol messages
+ if (CreateTransaction.isSerializedType(message)) {
+ handleCreateTransaction(message);
+ } else if (message instanceof BatchedModifications request) {
+ handleBatchedModifications(request);
+ } else if (message instanceof ForwardedReadyTransaction request) {
+ handleForwardedReadyTransaction(request);
+ } else if (message instanceof ReadyLocalTransaction request) {
+ handleReadyLocalTransaction(request);
+ } else if (CanCommitTransaction.isSerializedType(message)) {
+ handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
+ } else if (CommitTransaction.isSerializedType(message)) {
+ handleCommitTransaction(CommitTransaction.fromSerializable(message));
+ } else if (AbortTransaction.isSerializedType(message)) {
+ handleAbortTransaction(AbortTransaction.fromSerializable(message));
+ } else if (CloseTransactionChain.isSerializedType(message)) {
+ closeTransactionChain(CloseTransactionChain.fromSerializable(message));
+ } else if (ShardTransactionMessageRetrySupport.TIMER_MESSAGE_CLASS.isInstance(message)) {
+ messageRetrySupport.onTimerMessage(message);
+ } else {
+ super.handleNonRaftCommand(message);
+ }
}
}
}
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleCommitTransaction(final CommitTransaction commit) {
- final TransactionIdentifier txId = commit.getTransactionId();
+ final var txId = commit.getTransactionId();
if (isLeader()) {
askProtocolEncountered(txId);
commitCoordinator.handleCommit(txId, getSender(), this);
} else {
- ActorSelection leader = getLeader();
+ final var leader = getLeader();
if (leader == null) {
messageRetrySupport.addMessageToRetry(commit, getSender(), "Could not commit transaction " + txId);
} else {
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
- final TransactionIdentifier txId = canCommit.getTransactionId();
+ final var txId = canCommit.getTransactionId();
LOG.debug("{}: Can committing transaction {}", persistenceId(), txId);
if (isLeader()) {
askProtocolEncountered(txId);
commitCoordinator.handleCanCommit(txId, getSender(), this);
} else {
- ActorSelection leader = getLeader();
+ final var leader = getLeader();
if (leader == null) {
messageRetrySupport.addMessageToRetry(canCommit, getSender(),
"Could not canCommit transaction " + txId);
}
@SuppressWarnings("checkstyle:IllegalCatch")
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleBatchedModificationsLocal(final BatchedModifications batched, final ActorRef sender) {
askProtocolEncountered(batched.getTransactionId());
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleBatchedModifications(final BatchedModifications batched) {
// This message is sent to prepare the modifications transaction directly on the Shard as an
// optimization to avoid the extra overhead of a separate ShardTransaction actor. On the last
if (isLeader() && isLeaderActive) {
handleBatchedModificationsLocal(batched, getSender());
} else {
- ActorSelection leader = getLeader();
+ final var leader = getLeader();
if (!isLeaderActive || leader == null) {
messageRetrySupport.addMessageToRetry(batched, getSender(),
"Could not process BatchedModifications " + batched.getTransactionId());
// we need to reconstruct previous BatchedModifications from the transaction
// DataTreeModification, honoring the max batched modification count, and forward all the
// previous BatchedModifications to the new leader.
- Collection<BatchedModifications> newModifications = commitCoordinator
- .createForwardedBatchedModifications(batched,
- datastoreContext.getShardBatchedModificationCount());
+ final var newModifications = commitCoordinator.createForwardedBatchedModifications(batched,
+ datastoreContext.getShardBatchedModificationCount());
LOG.debug("{}: Forwarding {} BatchedModifications to leader {}", persistenceId(),
newModifications.size(), leader);
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void handleReadyLocalTransaction(final ReadyLocalTransaction message) {
- final TransactionIdentifier txId = message.getTransactionId();
+ @Deprecated(since = "9.0.0", forRemoval = true)
+ private void handleReadyLocalTransaction(final ReadyLocalTransaction message) {
+ final var txId = message.getTransactionId();
LOG.debug("{}: handleReadyLocalTransaction for {}", persistenceId(), txId);
- boolean isLeaderActive = isLeaderActive();
+ final var isLeaderActive = isLeaderActive();
if (isLeader() && isLeaderActive) {
askProtocolEncountered(txId);
try {
getSender().tell(new Failure(e), getSelf());
}
} else {
- ActorSelection leader = getLeader();
+ final var leader = getLeader();
if (!isLeaderActive || leader == null) {
messageRetrySupport.addMessageToRetry(message, getSender(),
"Could not process ready local transaction " + txId);
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleForwardedReadyTransaction(final ForwardedReadyTransaction forwardedReady) {
LOG.debug("{}: handleForwardedReadyTransaction for {}", persistenceId(), forwardedReady.getTransactionId());
- boolean isLeaderActive = isLeaderActive();
+ final var isLeaderActive = isLeaderActive();
if (isLeader() && isLeaderActive) {
askProtocolEncountered(forwardedReady.getTransactionId());
commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this);
} else {
- ActorSelection leader = getLeader();
+ final var leader = getLeader();
if (!isLeaderActive || leader == null) {
messageRetrySupport.addMessageToRetry(forwardedReady, getSender(),
"Could not process forwarded ready transaction " + forwardedReady.getTransactionId());
} else {
LOG.debug("{}: Forwarding ForwardedReadyTransaction to leader {}", persistenceId(), leader);
- ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionId(),
+ final var readyLocal = new ReadyLocalTransaction(forwardedReady.getTransactionId(),
forwardedReady.getTransaction().getSnapshot(), forwardedReady.isDoImmediateCommit(),
forwardedReady.getParticipatingShardNames());
readyLocal.setRemoteVersion(getCurrentBehavior().getLeaderPayloadVersion());
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleAbortTransaction(final AbortTransaction abort) {
- final TransactionIdentifier transactionId = abort.getTransactionId();
+ final var transactionId = abort.getTransactionId();
askProtocolEncountered(transactionId);
doAbortTransaction(transactionId, getSender());
}
commitCoordinator.handleAbort(transactionID, sender, this);
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void handleCreateTransaction(final Object message) {
if (isLeader()) {
createTransaction(CreateTransaction.fromSerializable(message));
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
if (isLeader()) {
- final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier();
+ final var id = closeTransactionChain.getIdentifier();
askProtocolEncountered(id.getClientId());
store.closeTransactionChain(id);
} else if (getLeader() != null) {
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
@SuppressWarnings("checkstyle:IllegalCatch")
private void createTransaction(final CreateTransaction createTransaction) {
askProtocolEncountered(createTransaction.getTransactionId());
return;
}
- ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
+ final var transactionActor = createTransaction(createTransaction.getTransactionType(),
createTransaction.getTransactionId());
getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
}
}
+ @Deprecated(since = "9.0.0", forRemoval = true)
private ActorRef createTransaction(final int transactionType, final TransactionIdentifier transactionId) {
LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId);
return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType),
}
// Called on leader only
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void askProtocolEncountered(final TransactionIdentifier transactionId) {
askProtocolEncountered(transactionId.getHistoryId().getClientId());
}
// Called on leader only
+ @Deprecated(since = "9.0.0", forRemoval = true)
private void askProtocolEncountered(final ClientIdentifier clientId) {
- final FrontendIdentifier frontend = clientId.getFrontendId();
- final LeaderFrontendState state = knownFrontends.get(frontend);
+ final var frontend = clientId.getFrontendId();
+ final var state = knownFrontends.get(frontend);
if (!(state instanceof LeaderFrontendState.Disabled)) {
LOG.debug("{}: encountered ask-based client {}, disabling transaction tracking", persistenceId(), clientId);
if (knownFrontends.isEmpty()) {
}
private void updateSchemaContext(final UpdateSchemaContext message) {
- updateSchemaContext(message.getEffectiveModelContext());
+ updateSchemaContext(message.modelContext());
}
@VisibleForTesting
private ShardIdentifier id;
private Map<String, String> peerAddresses = Collections.emptyMap();
private DatastoreContext datastoreContext;
- private EffectiveModelContextProvider schemaContextProvider;
+ private Supplier<@NonNull EffectiveModelContext> schemaContextProvider;
private DatastoreSnapshot.ShardSnapshot restoreFromSnapshot;
private DataTree dataTree;
return self();
}
- public T schemaContextProvider(final EffectiveModelContextProvider newSchemaContextProvider) {
+ public T schemaContextProvider(final Supplier<@NonNull EffectiveModelContext> newSchemaContextProvider) {
checkSealed();
schemaContextProvider = requireNonNull(newSchemaContextProvider);
return self();
}
public EffectiveModelContext getSchemaContext() {
- return verifyNotNull(schemaContextProvider.getEffectiveModelContext());
+ return verifyNotNull(schemaContextProvider.get());
}
public DatastoreSnapshot.ShardSnapshot getRestoreFromSnapshot() {
Code Review / controller.git / blobdiff