- @Override public void onReceiveCommand(Object message) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("onReceiveCommand: Received message {} from {}",
- message.getClass().toString(),
- getSender());
- }
-
- if(message.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
- // This must be for install snapshot. Don't want to open this up and trigger
- // deSerialization
- self()
- .tell(new CaptureSnapshotReply(ReadDataReply.getNormalizedNodeByteString(message)),
- self());
-
- createSnapshotTransaction = null;
- // Send a PoisonPill instead of sending close transaction because we do not really need
- // a response
- getSender().tell(PoisonPill.getInstance(), self());
-
- } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)){
- closeTransactionChain(CloseTransactionChain.fromSerializable(message));
- } else if (message instanceof RegisterChangeListener) {
- registerChangeListener((RegisterChangeListener) message);
- } else if (message instanceof UpdateSchemaContext) {
- updateSchemaContext((UpdateSchemaContext) message);
- } else if (message instanceof ForwardedCommitTransaction) {
- handleForwardedCommit((ForwardedCommitTransaction) message);
- } else if (message.getClass()
- .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
- if (isLeader()) {
- createTransaction(CreateTransaction.fromSerializable(message));
- } else if (getLeader() != null) {
- getLeader().forward(message, getContext());
+ @Override
+ public void onReceiveCommand(final Object message) throws Exception {
+
+ MessageTracker.Context context = appendEntriesReplyTracker.received(message);
+
+ if(context.error().isPresent()){
+ LOG.trace("{} : AppendEntriesReply failed to arrive at the expected interval {}", persistenceId(),
+ context.error());
+ }
+
+ try {
+ if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
+ handleCreateTransaction(message);
+ } else if (message instanceof ForwardedReadyTransaction) {
+ handleForwardedReadyTransaction((ForwardedReadyTransaction) message);
+ } else if (message.getClass().equals(CanCommitTransaction.SERIALIZABLE_CLASS)) {
+ handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
+ } else if (message.getClass().equals(CommitTransaction.SERIALIZABLE_CLASS)) {
+ handleCommitTransaction(CommitTransaction.fromSerializable(message));
+ } else if (message.getClass().equals(AbortTransaction.SERIALIZABLE_CLASS)) {
+ handleAbortTransaction(AbortTransaction.fromSerializable(message));
+ } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)) {
+ closeTransactionChain(CloseTransactionChain.fromSerializable(message));
+ } else if (message instanceof RegisterChangeListener) {
+ registerChangeListener((RegisterChangeListener) message);
+ } else if (message instanceof UpdateSchemaContext) {
+ updateSchemaContext((UpdateSchemaContext) message);
+ } else if (message instanceof PeerAddressResolved) {
+ PeerAddressResolved resolved = (PeerAddressResolved) message;
+ setPeerAddress(resolved.getPeerId().toString(),
+ resolved.getPeerAddress());
+ } else if (message.equals(TX_COMMIT_TIMEOUT_CHECK_MESSAGE)) {
+ handleTransactionCommitTimeoutCheck();
+ } else if(message instanceof DatastoreContext) {
+ onDatastoreContext((DatastoreContext)message);
+ } else {
+ super.onReceiveCommand(message);
+ }
+ } finally {
+ context.done();
+ }
+ }
+
+ @Override
+ protected Optional<ActorRef> getRoleChangeNotifier() {
+ return roleChangeNotifier;
+ }
+
+ private void onDatastoreContext(DatastoreContext context) {
+ datastoreContext = context;
+
+ commitCoordinator.setQueueCapacity(datastoreContext.getShardTransactionCommitQueueCapacity());
+
+ setTransactionCommitTimeout();
+
+ if(datastoreContext.isPersistent() &&
+ dataPersistenceProvider instanceof NonPersistentRaftDataProvider) {
+ dataPersistenceProvider = new PersistentDataProvider();
+ } else if(!datastoreContext.isPersistent() &&
+ dataPersistenceProvider instanceof PersistentDataProvider) {
+ dataPersistenceProvider = new NonPersistentRaftDataProvider();
+ }
+
+ updateConfigParams(datastoreContext.getShardRaftConfig());
+ }
+
+ private void handleTransactionCommitTimeoutCheck() {
+ CohortEntry cohortEntry = commitCoordinator.getCurrentCohortEntry();
+ if(cohortEntry != null) {
+ long elapsed = System.currentTimeMillis() - cohortEntry.getLastAccessTime();
+ if(elapsed > transactionCommitTimeout) {
+ LOG.warn("{}: Current transaction {} has timed out after {} ms - aborting",
+ persistenceId(), cohortEntry.getTransactionID(), transactionCommitTimeout);
+
+ doAbortTransaction(cohortEntry.getTransactionID(), null);
+ }
+ }
+ }
+
+ private void handleCommitTransaction(final CommitTransaction commit) {
+ final String transactionID = commit.getTransactionID();
+
+ LOG.debug("{}: Committing transaction {}", persistenceId(), transactionID);
+
+ // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
+ // this transaction.
+ final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
+ if(cohortEntry == null) {
+ // We're not the current Tx - the Tx was likely expired b/c it took too long in
+ // between the canCommit and commit messages.
+ IllegalStateException ex = new IllegalStateException(
+ String.format("%s: Cannot commit transaction %s - it is not the current transaction",
+ persistenceId(), transactionID));
+ LOG.error(ex.getMessage());
+ shardMBean.incrementFailedTransactionsCount();
+ getSender().tell(new akka.actor.Status.Failure(ex), getSelf());
+ return;
+ }
+
+ // We perform the preCommit phase here atomically with the commit phase. This is an
+ // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
+ // coordination of preCommit across shards in case of failure but preCommit should not
+ // normally fail since we ensure only one concurrent 3-phase commit.
+
+ try {
+ // We block on the future here so we don't have to worry about possibly accessing our
+ // state on a different thread outside of our dispatcher. Also, the data store
+ // currently uses a same thread executor anyway.
+ cohortEntry.getCohort().preCommit().get();
+
+ // If we do not have any followers and we are not using persistence we can
+ // apply modification to the state immediately
+ if(!hasFollowers() && !persistence().isRecoveryApplicable()){
+ applyModificationToState(getSender(), transactionID, cohortEntry.getModification());
+ } else {
+ Shard.this.persistData(getSender(), transactionID,
+ new ModificationPayload(cohortEntry.getModification()));
+ }
+ } catch (Exception e) {
+ LOG.error("{} An exception occurred while preCommitting transaction {}",
+ persistenceId(), cohortEntry.getTransactionID(), e);
+ shardMBean.incrementFailedTransactionsCount();
+ getSender().tell(new akka.actor.Status.Failure(e), getSelf());
+ }
+
+ cohortEntry.updateLastAccessTime();
+ }
+
+ private void finishCommit(@Nonnull final ActorRef sender, final @Nonnull String transactionID) {
+ // With persistence enabled, this method is called via applyState by the leader strategy
+ // after the commit has been replicated to a majority of the followers.
+
+ CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
+ if(cohortEntry == null) {
+ // The transaction is no longer the current commit. This can happen if the transaction
+ // was aborted prior, most likely due to timeout in the front-end. We need to finish
+ // committing the transaction though since it was successfully persisted and replicated
+ // however we can't use the original cohort b/c it was already preCommitted and may
+ // conflict with the current commit or may have been aborted so we commit with a new
+ // transaction.
+ cohortEntry = commitCoordinator.getAndRemoveCohortEntry(transactionID);
+ if(cohortEntry != null) {
+ commitWithNewTransaction(cohortEntry.getModification());
+ sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());