import akka.dispatch.OnComplete;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
+import java.util.Optional;
+import java.util.SortedSet;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
this.modification = null;
}
- private Future<Object> initiateCommit(final boolean immediate) {
+ private Future<Object> initiateCommit(final boolean immediate,
+ final Optional<SortedSet<String>> participatingShardNames) {
if (operationError != null) {
return Futures.failed(operationError);
}
final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier(),
- modification, immediate);
+ modification, immediate, participatingShardNames);
return actorContext.executeOperationAsync(leader, message, actorContext.getTransactionCommitOperationTimeout());
}
- Future<ActorSelection> initiateCoordinatedCommit() {
- final Future<Object> messageFuture = initiateCommit(false);
+ Future<ActorSelection> initiateCoordinatedCommit(final Optional<SortedSet<String>> participatingShardNames) {
+ final Future<Object> messageFuture = initiateCommit(false, participatingShardNames);
final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorContext,
transaction.getIdentifier());
ret.onComplete(new OnComplete<ActorSelection>() {
@Override
- public void onComplete(final Throwable failure, final ActorSelection success) throws Throwable {
+ public void onComplete(final Throwable failure, final ActorSelection success) {
if (failure != null) {
- LOG.info("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
+ LOG.warn("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
transactionAborted(transaction);
return;
}
}
Future<Object> initiateDirectCommit() {
- final Future<Object> messageFuture = initiateCommit(true);
+ final Future<Object> messageFuture = initiateCommit(true, Optional.empty());
messageFuture.onComplete(new OnComplete<Object>() {
@Override
- public void onComplete(final Throwable failure, final Object message) throws Throwable {
+ public void onComplete(final Throwable failure, final Object message) {
if (failure != null) {
- LOG.error("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
+ LOG.warn("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
transactionAborted(transaction);
} else if (CommitTransactionReply.isSerializedType(message)) {
LOG.debug("Transaction {} committed successfully", transaction.getIdentifier());
transactionCommitted(transaction);
} else {
- LOG.error("Transaction {} resulted in unhandled message type {}, aborting", message.getClass());
+ LOG.error("Transaction {} resulted in unhandled message type {}, aborting",
+ transaction.getIdentifier(), message.getClass());
transactionAborted(transaction);
}
}
throw new UnsupportedOperationException();
}
- protected void transactionAborted(SnapshotBackedWriteTransaction<TransactionIdentifier> aborted) {
+ protected void transactionAborted(final SnapshotBackedWriteTransaction<TransactionIdentifier> aborted) {
}
- protected void transactionCommitted(SnapshotBackedWriteTransaction<TransactionIdentifier> comitted) {
+ protected void transactionCommitted(final SnapshotBackedWriteTransaction<TransactionIdentifier> comitted) {
}
}