import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
-import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
+
+import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import javax.annotation.Nonnull;
/**
* A Shard represents a portion of the logical data tree <br/>
if (message instanceof RecoveryFailure){
LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause");
+
+ // Even though recovery failed, we still need to finish our recovery, eg send the
+ // ActorInitialized message and start the txCommitTimeoutCheckSchedule.
+ onRecoveryComplete();
} else {
super.onReceiveRecover(message);
}
ready.getModification());
// Return our actor path as we'll handle the three phase commit.
- getSender().tell(new ReadyTransactionReply(Serialization.serializedActorPath(self())).
- toSerializable(), getSelf());
+ ReadyTransactionReply readyTransactionReply =
+ new ReadyTransactionReply(Serialization.serializedActorPath(self()));
+ getSender().tell(
+ ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : readyTransactionReply,
+ getSelf());
}
private void handleAbortTransaction(AbortTransaction abort) {
//notify shard manager
getContext().parent().tell(new ActorInitialized(), getSelf());
- // Schedule a message to be periodically sent to check if the current in-progress
- // transaction should be expired and aborted.
- FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
- txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
- period, period, getSelf(),
- TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
+ // Being paranoid here - this method should only be called once but just in case...
+ if(txCommitTimeoutCheckSchedule == null) {
+ // Schedule a message to be periodically sent to check if the current in-progress
+ // transaction should be expired and aborted.
+ FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
+ txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
+ period, period, getSelf(),
+ TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
+ }
}
@Override