- /**
- * This method is overridden to ensure the previous Tx's ready operations complete
- * before we create the next shard Tx in the chain to avoid creation failures if the
- * previous Tx's ready operations haven't completed yet.
- */
- @Override
- protected Future<Object> sendCreateTransaction(final ActorSelection shard,
- final Object serializedCreateMessage) {
-
- // Check if there are any previous ready Futures, otherwise let the super class handle it.
- // The second check is done to ensure the the previous ready Futures aren't for this
- // Tx instance as deadlock would occur if we tried to wait on our own Futures. This can
- // occur in this scenario:
- //
- // - the TransactionProxy is created and the client does a write.
- //
- // - the TransactionProxy then attempts to create the shard Tx. However it first
- // sends a FindPrimaryShard message to the shard manager to find the local shard
- // This call is done async.
- //
- // - the client submits the Tx and the TransactionProxy is readied and we cache
- // the ready Futures here.
- //
- // - then the FindPrimaryShard call completes and this method is called to create
- // the shard Tx. However the cached Futures were from the ready on this Tx. If we
- // tried to wait on them, it would cause a form of deadlock as the ready Future
- // would be waiting on the Tx create Future and vice versa.
- SimpleEntry<Object, List<Future<ActorSelection>>> readyFuturesEntry = state.getReadyFutures();
- List<Future<ActorSelection>> readyFutures = readyFuturesEntry.getValue();
- if(readyFutures.isEmpty() || getIdentifier().equals(readyFuturesEntry.getKey())) {
- return super.sendCreateTransaction(shard, serializedCreateMessage);
+ previous = combineFutureWithPossiblePriorReadOnlyTxFutures(previous, txId);
+
+ // Add a callback for completion of the combined Futures.
+ final Promise<PrimaryShardInfo> returnPromise = Futures.promise();
+
+ final OnComplete onComplete = new OnComplete() {
+ @Override
+ public void onComplete(final Throwable failure, final Object notUsed) {
+ if (failure != null) {
+ // A Ready Future failed so fail the returned Promise.
+ LOG.error("Tx: {} - ready future failed for previous Tx {}", txId, previousTransactionId);
+ returnPromise.failure(failure);
+ } else {
+ LOG.debug("Tx: {} - previous Tx {} readied - proceeding to FindPrimaryShard",
+ txId, previousTransactionId);
+
+ // Send the FindPrimaryShard message and use the resulting Future to complete the
+ // returned Promise.
+ returnPromise.completeWith(parent.findPrimaryShard(shardName, txId));
+ }
+ }
+ };
+
+ previous.onComplete(onComplete, getActorContext().getClientDispatcher());
+ return returnPromise.future();
+ }
+
+ private <T> Future<T> combineFutureWithPossiblePriorReadOnlyTxFutures(final Future<T> future,
+ final TransactionIdentifier txId) {
+ if (!priorReadOnlyTxPromises.containsKey(txId) && !priorReadOnlyTxPromises.isEmpty()) {
+ Collection<Entry<TransactionIdentifier, Promise<Object>>> priorReadOnlyTxPromiseEntries =
+ new ArrayList<>(priorReadOnlyTxPromises.entrySet());
+ if (priorReadOnlyTxPromiseEntries.isEmpty()) {
+ return future;