@@ -374,7 +374,8 @@ public class Shard extends RaftActor { } private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) { - LOG.debug("Readying transaction {}", ready.getTransactionID()); + LOG.debug("Readying transaction {}, client version {}", ready.getTransactionID(), + ready.getTxnClientVersion()); // This message is forwarded by the ShardTransaction on ready. We cache the cohort in the // commitCoordinator in preparation for the subsequent three phase commit initiated by @@ -382,12 +383,22 @@ public class Shard extends RaftActor { commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(), ready.getModification()); - // Return our actor path as we'll handle the three phase commit. - ReadyTransactionReply readyTransactionReply = - new ReadyTransactionReply(Serialization.serializedActorPath(self())); - getSender().tell( - ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : readyTransactionReply, - getSelf()); + // Return our actor path as we'll handle the three phase commit, except if the Tx client + // version < 1 (Helium-1 version). This means the Tx was initiated by a base Helium version + // node. In that case, the subsequent 3-phase commit messages won't contain the + // transactionId so to maintain backwards compatibility, we create a separate cohort actor + // to provide the compatible behavior. + ActorRef replyActorPath = self(); + if(ready.getTxnClientVersion() < CreateTransaction.HELIUM_1_VERSION) { + LOG.debug("Creating BackwardsCompatibleThreePhaseCommitCohort"); + replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props( + ready.getTransactionID())); + } + + ReadyTransactionReply readyTransactionReply = new ReadyTransactionReply( + Serialization.serializedActorPath(replyActorPath)); + getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : + readyTransactionReply, getSelf()); } private void handleAbortTransaction(AbortTransaction abort) { @@ -465,10 +476,8 @@ public class Shard extends RaftActor { } } - private ActorRef createTypedTransactionActor( - int transactionType, - ShardTransactionIdentifier transactionId, - String transactionChainId ) { + private ActorRef createTypedTransactionActor(int transactionType, + ShardTransactionIdentifier transactionId, String transactionChainId, int clientVersion ) { DOMStoreTransactionFactory factory = store; @@ -481,8 +490,8 @@ public class Shard extends RaftActor { } } - if(this.schemaContext == null){ - throw new NullPointerException("schemaContext should not be null"); + if(this.schemaContext == null) { + throw new IllegalStateException("SchemaContext is not set"); } if (transactionType == TransactionProxy.TransactionType.READ_ONLY.ordinal()) { @@ -492,7 +501,8 @@ public class Shard extends RaftActor { return getContext().actorOf( ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(), schemaContext,datastoreContext, shardMBean, - transactionId.getRemoteTransactionId()), transactionId.toString()); + transactionId.getRemoteTransactionId(), clientVersion), + transactionId.toString()); } else if (transactionType == TransactionProxy.TransactionType.READ_WRITE.ordinal()) { @@ -501,7 +511,8 @@ public class Shard extends RaftActor { return getContext().actorOf( ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(), schemaContext, datastoreContext, shardMBean, - transactionId.getRemoteTransactionId()), transactionId.toString()); + transactionId.getRemoteTransactionId(), clientVersion), + transactionId.toString()); } else if (transactionType == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) { @@ -511,7 +522,8 @@ public class Shard extends RaftActor { return getContext().actorOf( ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(), schemaContext, datastoreContext, shardMBean, - transactionId.getRemoteTransactionId()), transactionId.toString()); + transactionId.getRemoteTransactionId(), clientVersion), + transactionId.toString()); } else { throw new IllegalArgumentException( "Shard="+name + ":CreateTransaction message has unidentified transaction type=" @@ -520,27 +532,32 @@ public class Shard extends RaftActor { } private void createTransaction(CreateTransaction createTransaction) { - createTransaction(createTransaction.getTransactionType(), - createTransaction.getTransactionId(), createTransaction.getTransactionChainId()); + try { + ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(), + createTransaction.getTransactionId(), createTransaction.getTransactionChainId(), + createTransaction.getVersion()); + + getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), + createTransaction.getTransactionId()).toSerializable(), getSelf()); + } catch (Exception e) { + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } - private ActorRef createTransaction(int transactionType, String remoteTransactionId, String transactionChainId) { + private ActorRef createTransaction(int transactionType, String remoteTransactionId, + String transactionChainId, int clientVersion) { ShardTransactionIdentifier transactionId = ShardTransactionIdentifier.builder() .remoteTransactionId(remoteTransactionId) .build(); + if(LOG.isDebugEnabled()) { LOG.debug("Creating transaction : {} ", transactionId); } - ActorRef transactionActor = - createTypedTransactionActor(transactionType, transactionId, transactionChainId); - getSender() - .tell(new CreateTransactionReply( - Serialization.serializedActorPath(transactionActor), - remoteTransactionId).toSerializable(), - getSelf()); + ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId, + transactionChainId, clientVersion); return transactionActor; } @@ -599,7 +616,7 @@ public class Shard extends RaftActor { LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = {} ", listenerRegistration.path()); - getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()),getSelf()); + getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf()); } private ListenerRegistration