Two Part Commit

Locks and Transaction Isolation

The requests additionally take a lock on the keys.
Notably, the get requests take a learn lock and
the put requests take a write lock. The learn locks are taken because the
values are learn.

class TransactionalKVStore…

  public CompletableFuture<String> get(TransactionRef txn, String key) {
      CompletableFuture<TransactionRef> lockFuture
              = lockManager.purchase(txn, key, LockMode.READ);
      return lockFuture.thenApply(transactionRef -> {
          getOrCreateTransactionState(transactionRef);
          return kv.get(key);
      });
  }

  synchronized TransactionState getOrCreateTransactionState(TransactionRef txnRef) {
      TransactionState state = this.ongoingTransactions.get(txnRef);
      if (state == null) {
          state = new TransactionState();
          this.ongoingTransactions.put(txnRef, state);
      }
      return state;
  }

The write locks might be taken solely when the transaction is about to commit
and the values are to be made seen in the important thing worth retailer. Till then, the
cluster node can simply monitor the modified values as pending operations.

Delaying locking decreases the possibilities of conflicting transactions.

class TransactionalKVStore…

  public void put(TransactionRef transactionRef, String key, String worth) {
      TransactionState state = getOrCreateTransactionState(transactionRef);
      state.addPendingUpdates(key, worth);
  }

You will need to be aware that the locks are lengthy lived and never launched
when the request completes. They’re launched solely when the transaction commits.
This method of holding locks at some stage in the transaction
and releasing them solely when the transaction commits or rolls again is
referred to as two-phase-locking.
Two-phase locking is essential in offering the serializable isolation degree.
Serializable which means that the consequences of the transactions are seen as
if they’re executed one after the other.

  WaitPolicy waitPolicy;

public enum WaitPolicy {
    WoundWait,
    WaitDie,
    Error
}

  Queue<LockRequest> waitQueue = new LinkedList<>();
  Checklist<TransactionRef> house owners = new ArrayList<>();
  LockMode lockMode;

  public synchronized CompletableFuture<TransactionRef> purchase(TransactionRef txn, String key, LockMode lockMode) {
      return purchase(txn, key, lockMode, new CompletableFuture<>());
  }

  CompletableFuture<TransactionRef> purchase(TransactionRef txnRef,
                                            String key,
                                            LockMode askedLockMode,
                                            CompletableFuture<TransactionRef> lockFuture) {
      Lock lock = getOrCreateLock(key);

      logger.debug("buying lock for = " + txnRef + " on key = " + key + " with lock mode = " + askedLockMode);
      if (lock.isCompatible(txnRef, askedLockMode)) {
          lock.addOwner(txnRef, askedLockMode);
          lockFuture.full(txnRef);
          logger.debug("acquired lock for = " + txnRef);
          return lockFuture;
      }

  public boolean isCompatible(TransactionRef txnRef, LockMode lockMode) {
      if(hasOwner()) 
      return true;
  }

  personal CompletableFuture<TransactionRef> handleConflict(Lock lock,
                                                           TransactionRef txnRef,
                                                           String key,
                                                           LockMode askedLockMode,
                                                           CompletableFuture<TransactionRef> lockFuture) {
      swap (waitPolicy) {
          case Error: {
              lockFuture.completeExceptionally(new WriteConflictException(txnRef, key, lock.house owners));
              return lockFuture;
          }
          case WoundWait: {
              return lock.woundWait(txnRef, key, askedLockMode, lockFuture, this);
          }
          case WaitDie: {
              return lock.waitDie(txnRef, key, askedLockMode, lockFuture, this);
          }
      }
      throw new IllegalArgumentException("Unknown waitPolicy " + waitPolicy);
  }

  boolean after(TransactionRef otherTransactionRef) {
      return this.startTimestamp > otherTransactionRef.startTimestamp;
  }

  personal int requestId;
  public MonotonicId start() {
      return new MonotonicId(requestId++, config.getServerId());
  }

  public class MonotonicId implements Comparable<MonotonicId> {
      public int requestId;
      int serverId;
  
      public MonotonicId(int requestId, int serverId) {
          this.serverId = serverId;
          this.requestId = requestId;
      }
  
      public static MonotonicId empty() {
          return new MonotonicId(-1, -1);
      }
  
      public boolean isAfter(MonotonicId different) {
          if (this.requestId == different.requestId) {
              return this.serverId > different.serverId;
          }
          return this.requestId > different.requestId;
      }

  personal void beginTransaction(String key) {
      if (coordinator == null) {
          coordinator = replicaMapper.serverFor(key);
          MonotonicId transactionId = coordinator.start();
          transactionRef = new TransactionRef(transactionId, clock.nanoTime());
      }
  }

  public void incrementAge(SystemClock clock) {
      age = clock.nanoTime() - startTimestamp;
  }

  public boolean isAfter(TransactionRef different) {
       return age == different.age?
                  this.id.isAfter(different.id)
                  :this.age > different.age;
  }

  public CompletableFuture<TransactionRef> woundWait(TransactionRef txnRef,
                                                     String key,
                                                     LockMode askedLockMode,
                                                     CompletableFuture<TransactionRef> lockFuture,
                                                     LockManager lockManager) {

      if (allOwningTransactionsStartedAfter(txnRef) && !anyOwnerIsPrepared(lockManager)) {
          abortAllOwners(lockManager, key, txnRef);
          return lockManager.purchase(txnRef, key, askedLockMode, lockFuture);
      }

      LockRequest lockRequest = new LockRequest(txnRef, key, askedLockMode, lockFuture);
      lockManager.logger.debug("Including to attend queue = " + lockRequest);
      addToWaitQueue(lockRequest);
      return lockFuture;
  }

  personal boolean allOwningTransactionsStartedAfter(TransactionRef txn) {
      return house owners.stream().filter(o -> !o.equals(txn)).allMatch(proprietor -> proprietor.after(txn));
  }

  public CompletableFuture<TransactionRef> waitDie(TransactionRef txnRef,
                                                   String key,
                                                   LockMode askedLockMode,
                                                   CompletableFuture<TransactionRef> lockFuture,
                                                   LockManager lockManager) {
      if (allOwningTransactionsStartedAfter(txnRef)) {
          addToWaitQueue(new LockRequest(txnRef, key, askedLockMode, lockFuture));
          return lockFuture;
      }

      lockManager.abort(txnRef, key);
      lockFuture.completeExceptionally(new WriteConflictException(txnRef, key, house owners));
      return lockFuture;
  }

  personal void launch(TransactionRef txn, String key) {
      Non-compulsory<Lock> lock = getLock(key);
      lock.ifPresent(l -> {
          l.launch(txn, this);
      });
  }

  public void launch(TransactionRef txn, LockManager lockManager) {
      removeOwner(txn);
      if (hasWaiters()) {
          LockRequest lockRequest = getFirst(lockManager.waitPolicy);
          lockManager.purchase(lockRequest.txn, lockRequest.key, lockRequest.lockMode, lockRequest.future);
      }
  }

Commit and Rollback

As soon as the consumer efficiently reads with out dealing with any conflicts and
writes all the important thing values, it initiates the commit request by sending
a commit request to the coordinator.

class TransactionClient…

  public CompletableFuture<Boolean> commit() {
      return coordinator.commit(transactionRef);
  }

The transaction coordinator data the state of the transaction as
making ready to commit. The coordinator implements the commit dealing with in
two phases.

• It first sends the put together request to every of the contributors.
• As soon as it receives a profitable response from all of the contributors,
  the coordinator marks the transaction as ready to finish.
  Then it sends the commit request to all of the contributors.

class TransactionCoordinator…

  public CompletableFuture<Boolean> commit(TransactionRef transactionRef)  {
      TransactionMetadata metadata = transactions.get(transactionRef);
      metadata.markPreparingToCommit(transactionLog);
      Checklist<CompletableFuture<Boolean>> allPrepared = sendPrepareRequestToParticipants(transactionRef);
      CompletableFuture<Checklist<Boolean>> futureList = sequence(allPrepared);
      return futureList.thenApply(outcome -> {
          if (!outcome.stream().allMatch(r -> r)) {
              logger.information("Rolling again = " + transactionRef);
              rollback(transactionRef);
              return false;
          }
          metadata.markPrepared(transactionLog);
          sendCommitMessageToParticipants(transactionRef);
          metadata.markCommitComplete(transactionLog);
          return true;
      });
  }

  public Checklist<CompletableFuture<Boolean>> sendPrepareRequestToParticipants(TransactionRef transactionRef)  {
      TransactionMetadata transactionMetadata = transactions.get(transactionRef);
      var transactionParticipants = getParticipants(transactionMetadata.getParticipatingKeys());
      return transactionParticipants.keySet()
              .stream()
              .map(server -> server.handlePrepare(transactionRef))
              .accumulate(Collectors.toList());
  }

  personal void sendCommitMessageToParticipants(TransactionRef transactionRef) {
      TransactionMetadata transactionMetadata = transactions.get(transactionRef);
      var participantsForKeys = getParticipants(transactionMetadata.getParticipatingKeys());
      participantsForKeys.keySet().stream()
              .forEach(kvStore -> {
                  Checklist<String> keys = participantsForKeys.get(kvStore);
                  kvStore.handleCommit(transactionRef, keys);
              });
  }

  personal Map<TransactionalKVStore, Checklist<String>> getParticipants(Checklist<String> participatingKeys) {
      return participatingKeys.stream()
              .map(okay -> Pair.of(serverFor(okay), okay))
              .accumulate(Collectors.groupingBy(Pair::getKey, Collectors.mapping(Pair::getValue, Collectors.toList())));
  }

The cluster node receiving the put together requests do two issues:

• It tries to seize the write locks for the entire keys.
• As soon as profitable, it writes the entire modifications to the write-ahead log.

If it could efficiently do these,
it could assure that there are not any conflicting transactions,
and even within the case of a crash the cluster node can recuperate all of the
required state to finish the transaction.

class TransactionalKVStore…

  public synchronized CompletableFuture<Boolean> handlePrepare(TransactionRef txn) {
      strive {
          TransactionState state = getTransactionState(txn);
          if (state.isPrepared()) {
              return CompletableFuture.completedFuture(true); //already ready.
          }

          if (state.isAborted()) {
              return CompletableFuture.completedFuture(false); //aborted by one other transaction.
          }

          Non-compulsory<Map<String, String>> pendingUpdates = state.getPendingUpdates();
          CompletableFuture<Boolean> prepareFuture = prepareUpdates(txn, pendingUpdates);
          return prepareFuture.thenApply(ignored -> {
              Map<String, Lock> locksHeldByTxn = lockManager.getAllLocksFor(txn);
              state.markPrepared();
              writeToWAL(new TransactionMarker(txn, locksHeldByTxn, TransactionStatus.PREPARED));
              return true;
          });

      } catch (TransactionException| WriteConflictException e) {
          logger.error(e);
      }
      return CompletableFuture.completedFuture(false);
  }

  personal CompletableFuture<Boolean> prepareUpdates(TransactionRef txn, Non-compulsory<Map<String, String>> pendingUpdates)  {
      if (pendingUpdates.isPresent()) {
          Map<String, String> pendingKVs = pendingUpdates.get();
          CompletableFuture<Checklist<TransactionRef>> lockFuture = acquireLocks(txn, pendingKVs.keySet());
          return lockFuture.thenApply(ignored -> {
              writeToWAL(txn, pendingKVs);
              return true;
          });
      }
      return CompletableFuture.completedFuture(true);
  }

  TransactionState getTransactionState(TransactionRef txnRef) {
      return ongoingTransactions.get(txnRef);
  }

  personal void writeToWAL(TransactionRef txn, Map<String, String> pendingUpdates) {
     for (String key : pendingUpdates.keySet()) {
          String worth = pendingUpdates.get(key);
          wal.writeEntry(new SetValueCommand(txn, key, worth).serialize());
      }
  }

  personal CompletableFuture<Checklist<TransactionRef>> acquireLocks(TransactionRef txn, Set<String> keys) {
      Checklist<CompletableFuture<TransactionRef>> lockFutures = new ArrayList<>();
      for (String key : keys) {
          CompletableFuture<TransactionRef> lockFuture = lockManager.purchase(txn, key, LockMode.READWRITE);
          lockFutures.add(lockFuture);
      }
      return sequence(lockFutures);
  }

When the cluster node receives the commit message from the coordinator,
it’s protected to make the key-value modifications seen.
The cluster node does three issues whereas committing the modifications:

• It marks the transaction as dedicated. Ought to the cluster node fail at this level,
  it is aware of the end result of the transaction, and might repeat the next steps.
• It applies all of the modifications to the key-value storage
• It releases all of the acquired locks.

class TransactionalKVStore…

  public synchronized void handleCommit(TransactionRef transactionRef, Checklist<String> keys) {
      if (!ongoingTransactions.containsKey(transactionRef)) {
          return; //it is a no-op. Already dedicated.
      }

      if (!lockManager.hasLocksFor(transactionRef, keys)) {
          throw new IllegalStateException("Transaction " + transactionRef + " ought to maintain all of the required locks for keys " + keys);
      }

      writeToWAL(new TransactionMarker(transactionRef, TransactionStatus.COMMITTED, keys));

      applyPendingUpdates(transactionRef);

      releaseLocks(transactionRef, keys);
  }

  personal void removeTransactionState(TransactionRef txnRef) {
      ongoingTransactions.take away(txnRef);
  }


  personal void applyPendingUpdates(TransactionRef txnRef) {
      TransactionState state = getTransactionState(txnRef);
      Non-compulsory<Map<String, String>> pendingUpdates = state.getPendingUpdates();
      apply(txnRef, pendingUpdates);
  }

  personal void apply(TransactionRef txnRef, Non-compulsory<Map<String, String>> pendingUpdates) {
      if (pendingUpdates.isPresent()) {
          Map<String, String> pendingKv = pendingUpdates.get();
          apply(pendingKv);
      }
      removeTransactionState(txnRef);
  }

  personal void apply(Map<String, String> pendingKv) {
      for (String key : pendingKv.keySet()) {
          String worth = pendingKv.get(key);
          kv.put(key, worth);
      }
  }
  personal void releaseLocks(TransactionRef txn, Checklist<String> keys) {
          lockManager.launch(txn, keys);
  }

  personal Lengthy writeToWAL(TransactionMarker transactionMarker) {
     return wal.writeEntry(transactionMarker.serialize());
  }

The rollback is carried out in an analogous means. If there’s any failure,
the consumer communicates with the coordinator to rollback the transaction.

class TransactionClient…

  public void rollback() {
      coordinator.rollback(transactionRef);
  }

The transaction coordinator data the state of the transaction as making ready
to rollback. Then it forwards the rollback request to the entire servers
which saved the values for the given transaction.
As soon as the entire requests are profitable, the coordinator marks the transaction
rollback as full.In case the coordinator crashes after the transaction
is marked as ‘ready to rollback’, it could carry on sending the rollback
messages to all of the collaborating cluster nodes.

class TransactionCoordinator…

  public void rollback(TransactionRef transactionRef) {
      transactions.get(transactionRef).markPrepareToRollback(this.transactionLog);

      sendRollbackMessageToParticipants(transactionRef);

      transactions.get(transactionRef).markRollbackComplete(this.transactionLog);
  }

  personal void sendRollbackMessageToParticipants(TransactionRef transactionRef) {
      TransactionMetadata transactionMetadata = transactions.get(transactionRef);
      var contributors = getParticipants(transactionMetadata.getParticipatingKeys());
      for (TransactionalKVStore kvStore : contributors.keySet()) {
          Checklist<String> keys = contributors.get(kvStore);
          kvStore.handleRollback(transactionMetadata.getTxn(), keys);
      }
  }

The cluster nodes receiving the rollback request does three issues:

• It data the state of the transaction as rolled again within the write-ahead log.
• It discards the transaction state.
• It releases the entire locks

class TransactionalKVStore…

  public synchronized void handleRollback(TransactionRef transactionRef, Checklist<String> keys) {
      if (!ongoingTransactions.containsKey(transactionRef)) {
          return; //no-op. Already rolled again.
      }
      writeToWAL(new TransactionMarker(transactionRef, TransactionStatus.ROLLED_BACK, keys));
      this.ongoingTransactions.take away(transactionRef);
      this.lockManager.launch(transactionRef, keys);
  }

An Instance State of affairs

Atomic Writes

Take into account the next situation. Paula Blue has a truck and Steven Inexperienced
has a backhoe.
The provision and the reserving standing of the truck and the backhoe
are saved on a distributed key-value retailer.
Relying on how the keys are mapped to servers,
Blue’s truck and Inexperienced’s backhoe bookings are saved on
separate cluster nodes.
Alice is making an attempt to ebook a truck
and backhoe for the development work she is planning to begin on a Monday.
She wants each the truck and the backhoe to be accessible.

The reserving situation occurs as follows.

Alice checks the supply of Blue’s truck and Inexperienced’s backhoe.
by studying the keys ‘truck_booking_monday’ and ‘backhoe_booking_monday’

If the values are empty, the reserving is free.
She reserves the truck and the backhoe.
It will be important that each the values are set atomically.
If there’s any failure, then not one of the values is about.

The commit occurs in two phases. The primary server Alice
contacts acts because the coordinator and executes the 2 phases.

The coordinator is a separate participant within the
protocol, and is proven that means on the sequence diagram. Nevertheless
often one of many servers (Blue or Inexperienced) acts as
the coordinator, thus taking part in two roles within the interplay.

Conflicting Transactions

Take into account a situation the place one other particular person, Bob, can also be making an attempt to ebook a
truck and backhoe for building work on the identical Monday.

The reserving situation occurs as follows:

• Each Alice and Bob learn the keys ‘truck_booking_monday’
  and ‘backhoe_booking_monday’
• Each see that the values are empty, which means the reserving is free.
• Each attempt to ebook the truck and the backhoe.

The expectation is that, solely Alice or Bob, ought to be capable of ebook,
as a result of the transactions are conflicting.
In case of errors, the entire movement must be retried and hopefully,
one will go forward with the reserving.
However in no state of affairs, ought to reserving be carried out partially.
Both each bookings must be carried out or neither is finished.

To test the supply, each Alice and Bob begin a transaction
and speak to Blue and Inexperienced’s servers respectively to test for the supply.
Blue holds a learn lock for the important thing “truck_booking_on_monday” and
Inexperienced holds a learn lock for the important thing “backhoe_booking_on_monday”.
As a result of learn locks are shared, each Alice and Bob can learn the values.

Alice and Bob see that each the bookings can be found on Monday.
So that they reserve by sending the put requests to servers.
Each the servers maintain the put requests within the short-term storage.

When Alice and Bob determine to commit the transactions-
assuming that Blue acts as a coordinator- it triggers the two-phase
commit protocol and sends the put together requests to itself and Inexperienced.

For Alice’s request it tries to seize a write lock for the important thing ‘truck_booking_on_monday’, which
it can’t get, as a result of there’s a conflicting learn lock grabbed by
one other transaction. So Alice’s transaction fails within the put together part.
The identical factor occurs with Bob’s request.

Transactions might be retried with a retry loop as follows:

class TransactionExecutor…

  public boolean executeWithRetry(Operate<TransactionClient, Boolean> txnMethod, ReplicaMapper replicaMapper, SystemClock systemClock) {
      for (int try = 1; try <= maxRetries; try++) {
          TransactionClient consumer = new TransactionClient(replicaMapper, systemClock);
          strive {
              boolean checkPassed = txnMethod.apply(consumer);
              Boolean successfullyCommitted = consumer.commit().get();
              return checkPassed && successfullyCommitted;
          } catch (Exception e) {
              logger.error("Write battle detected whereas executing." + consumer.transactionRef + " Retrying try " + try);
              consumer.rollback();
              randomWait(); //watch for random interval
          }

      }
      return false;
  }

The instance reserving code for Alice and Bob will look as follows:

class TransactionalKVStoreTest…

  @Take a look at
  public void retryWhenConflict()  (!aliceTxn.isSuccess() && bobTxn.isSuccess()), "ready for one txn to finish", Length.ofSeconds(50));
  

  personal TransactionExecutor bookTransactionally(Checklist<TransactionalKVStore> allServers, String person, SystemClock systemClock) {
      Checklist<String> bookingKeys = Arrays.asList("truck_booking_on_monday", "backhoe_booking_on_monday");
      TransactionExecutor t1 = new TransactionExecutor(allServers);
      t1.executeAsyncWithRetry(txnClient -> {
          if (txnClient.isAvailable(bookingKeys)) {
              txnClient.reserve(bookingKeys, person);
              return true;
          }
          return false;
      }, systemClock);
      return t1;
  }

On this case one of many transactions will finally succeed and
the opposite will again out.

Whereas it is rather simple to implement, with Error WaitPolicy ,
there will likely be a number of transaction restarts,decreasing the general
throughput.
As defined within the above part, if Wound-Wait coverage is used
it’s going to have fewer transaction restarts. Within the above instance,
just one transaction will presumably restart as a substitute of each restarting
in case of conflicts.

Utilizing Versioned Worth

It is rather constraining to have conflicts for all of the learn and write
operations, significantly so when the transactions might be read-only.
It’s optimum if read-only transactions can work with out holding any
locks and nonetheless assure that the values learn in a transaction
don’t change with a concurrent read-write transaction.

Information-stores typically retailer a number of variations of the values,
as described in Versioned Worth.
The model used is the timestamp following Lamport Clock.
Principally a Hybrid Clock is utilized in databases like
MongoDB or CockroachDB.
To make use of it with the two-phase commit protocol, the trick is that each server
collaborating within the transaction sends the timestamp it could write the
values at, as response to the put together request.
The coordinator chooses the utmost of those timestamps as a
commit timestamp and sends it together with the worth.
The collaborating servers then save the worth on the commit timestamp.
This permits read-only requests to be executed with out holding locks,
as a result of it is assured that the worth written at a specific timestamp
isn’t going to alter.

Take into account a easy instance as follows. Philip is working a report back to learn
the entire bookings that occurred till timestamp 2. If it’s a long-running
operation holding a lock, Alice, who’s making an attempt to ebook a truck, will likely be blocked
till Philip’s work completes. With Versioned Worth
Philip’s get requests, that are a part of a read-only operation, can proceed
at timestamp 2, whereas Alice’s reserving continues at timestamp 4.

Observe that learn requests that are a part of a read-write transaction,
nonetheless want to carry a lock.

The instance code with Lamport Clock seems as follows:

class MvccTransactionalKVStore…

  public String readOnlyGet(String key, lengthy readTimestamp) {
      adjustServerTimestamp(readTimestamp);
      return kv.get(new VersionedKey(key, readTimestamp));
  }

  public CompletableFuture<String> get(TransactionRef txn, String key, lengthy readTimestamp) {
      adjustServerTimestamp(readTimestamp);
      CompletableFuture<TransactionRef> lockFuture = lockManager.purchase(txn, key, LockMode.READ);
      return lockFuture.thenApply(transactionRef -> {
          getOrCreateTransactionState(transactionRef);
          return kv.get(key);
      });
  }

  personal void adjustServerTimestamp(lengthy readTimestamp) {
      this.timestamp = readTimestamp > this.timestamp ? readTimestamp:timestamp;
  }

  public void put(TransactionRef txnId, String key, String worth) {
      timestamp = timestamp + 1;
      TransactionState transactionState = getOrCreateTransactionState(txnId);
      transactionState.addPendingUpdates(key, worth);
  }

class MvccTransactionalKVStore…

  personal lengthy put together(TransactionRef txn, Non-compulsory<Map<String, String>> pendingUpdates) throws WriteConflictException, IOException {
      if (pendingUpdates.isPresent()) {
          Map<String, String> pendingKVs = pendingUpdates.get();

          acquireLocks(txn, pendingKVs);

          timestamp = timestamp + 1; //increment the timestamp for write operation.

          writeToWAL(txn, pendingKVs, timestamp);
       }
      return timestamp;
  }

class MvccTransactionCoordinator…

  public lengthy commit(TransactionRef txn) {
          lengthy commitTimestamp = put together(txn);

          TransactionMetadata transactionMetadata = transactions.get(txn);
          transactionMetadata.markPreparedToCommit(commitTimestamp, this.transactionLog);

          sendCommitMessageToAllTheServers(txn, commitTimestamp, transactionMetadata.getParticipatingKeys());

          transactionMetadata.markCommitComplete(transactionLog);

          return commitTimestamp;
  }


  public lengthy put together(TransactionRef txn) throws WriteConflictException {
      TransactionMetadata transactionMetadata = transactions.get(txn);
      Map<MvccTransactionalKVStore, Checklist<String>> keysToServers = getParticipants(transactionMetadata.getParticipatingKeys());
      Checklist<Lengthy> prepareTimestamps = new ArrayList<>();
      for (MvccTransactionalKVStore retailer : keysToServers.keySet()) {
          Checklist<String> keys = keysToServers.get(retailer);
          lengthy prepareTimestamp = retailer.put together(txn, keys);
          prepareTimestamps.add(prepareTimestamp);
      }
      return prepareTimestamps.stream().max(Lengthy::evaluate).orElse(txn.getStartTimestamp());
  }

All of the collaborating cluster nodes then retailer the key-values on the
commit timestamp.

class MvccTransactionalKVStore…

  public void commit(TransactionRef txn, Checklist<String> keys, lengthy commitTimestamp) {
      if (!lockManager.hasLocksFor(txn, keys)) {
          throw new IllegalStateException("Transaction ought to maintain all of the required locks");
      }

      adjustServerTimestamp(commitTimestamp);

      applyPendingOperations(txn, commitTimestamp);

      lockManager.launch(txn, keys);

      logTransactionMarker(new TransactionMarker(txn, TransactionStatus.COMMITTED, commitTimestamp, keys, Collections.EMPTY_MAP));
  }

  personal void applyPendingOperations(TransactionRef txnId, lengthy commitTimestamp) {
      Non-compulsory<TransactionState> transactionState = getTransactionState(txnId);
      if (transactionState.isPresent()) {
          TransactionState t = transactionState.get();
          Non-compulsory<Map<String, String>> pendingUpdates = t.getPendingUpdates();
          apply(txnId, pendingUpdates, commitTimestamp);
      }
  }

  personal void apply(TransactionRef txnId, Non-compulsory<Map<String, String>> pendingUpdates, lengthy commitTimestamp) {
      if (pendingUpdates.isPresent()) {
          Map<String, String> pendingKv = pendingUpdates.get();
          apply(pendingKv, commitTimestamp);
      }
      ongoingTransactions.take away(txnId);
  }


  personal void apply(Map<String, String> pendingKv, lengthy commitTimestamp) {
      for (String key : pendingKv.keySet()) {
          String worth = pendingKv.get(key);
          kv.put(new VersionedKey(key, commitTimestamp), worth);
      }
  }

Utilizing Replicated Log

To enhance fault tolerance cluster nodes use Replicated Log.
The coordinator makes use of Replicated Log to retailer the transaction log entries.

Contemplating the instance of Alice and Bob within the above part,
the Blue servers will likely be a gaggle of servers, so are the Inexperienced servers.
All of the reserving knowledge will likely be replicated throughout a set of servers.
Every request which is a part of the two-phase commit goes to the chief
of the server group. The replication is carried out utilizing
Replicated Log.

The consumer communicates with the chief of every server group.
The replication is important solely when the consumer decides to commit the
transaction, so it occurs as a part of the put together request.

The coordinator replicates each state change to replicated log as properly.

In a distributed datastore, every cluster node handles a number of partitions.
A Replicated Log is maintained per partition.
When Raft is used as a part of replication it is generally
known as multi-raft.

Shopper communicates with the chief of every partition collaborating in
the transaction.

Failure Dealing with

Two-phase commit protocol closely depends on the coordinator node
to speak the end result of the transaction.
Till the end result of the transaction is understood,
the person cluster nodes can’t permit another transactions
to jot down to the keys collaborating within the pending transaction.
The cluster nodes block till the end result of the transaction is understood.
This places some essential necessities on the coordinator

The coordinator wants to recollect the state of the transactions
even in case of a course of crash.

Coordinator makes use of Write-Forward Log to file each replace
to the state of the transaction.
This fashion, when the coordinator crashes and comes again up,
it could proceed to work on the transactions that are incomplete.

class TransactionCoordinator…

  public void loadTransactionsFromWAL() throws IOException {
      Checklist<WALEntry> walEntries = this.transactionLog.readAll();
      for (WALEntry walEntry : walEntries) {
          TransactionMetadata txnMetadata = (TransactionMetadata) Command.deserialize(new ByteArrayInputStream(walEntry.getData()));
          transactions.put(txnMetadata.getTxn(), txnMetadata);
      }
      startTransactionTimeoutScheduler();
      completePreparedTransactions();
  }
  personal void completePreparedTransactions() throws IOException {
      Checklist<Map.Entry<TransactionRef, TransactionMetadata>> preparedTransactions
              = transactions.entrySet().stream().filter(entry -> entry.getValue().isPrepared()).accumulate(Collectors.toList());
      for (Map.Entry<TransactionRef, TransactionMetadata> preparedTransaction : preparedTransactions) {
          TransactionMetadata txnMetadata = preparedTransaction.getValue();
          sendCommitMessageToParticipants(txnMetadata.getTxn());
      }
  }

The consumer can fail earlier than sending the commit message to the coordinator.

The transaction coordinator tracks when every transaction state was up to date.
If no state replace is acquired in a timeout interval, which is configured,
it triggers a transaction rollback.

class TransactionCoordinator…

  personal ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1);
  personal ScheduledFuture<?> taskFuture;
  personal lengthy transactionTimeoutMs = Lengthy.MAX_VALUE; //for now.

  public void startTransactionTimeoutScheduler() {
      taskFuture = scheduler.scheduleAtFixedRate(() -> timeoutTransactions(),
              transactionTimeoutMs,
              transactionTimeoutMs,
              TimeUnit.MILLISECONDS);
  }

  personal void timeoutTransactions() {
      for (TransactionRef txnRef : transactions.keySet()) {
          TransactionMetadata transactionMetadata = transactions.get(txnRef);
          lengthy now = systemClock.nanoTime();
          if (transactionMetadata.hasTimedOut(now)) {
              sendRollbackMessageToParticipants(transactionMetadata.getTxn());
              transactionMetadata.markRollbackComplete(transactionLog);
          }
      }
  }

Transactions throughout heterogenous techniques

The answer outlined right here demonstrates the two-phase commit implementation
in a homogenous system. Homogenous which means all of the cluster nodes are half
of the identical system and retailer identical form of knowledge. For instance
a distributed knowledge retailer like MongoDb or a distributed message dealer
like Kafka.

Traditionally, two-phase commit was principally mentioned within the context of
heterogeneous techniques. Most typical utilization of two-phase commit was
with [XA] transactions. Within the J2EE servers, it is rather
frequent to make use of two-phase commit throughout a message dealer and a database.
The commonest utilization sample is when a message must be produced
on a message dealer like ActiveMQ or JMS and a file must be
inserted/up to date in a database.

As seen within the above sections, the fault tolerance of the coordinator
performs a essential position in two-phase commit implementation. In case of XA
transactions the coordinator is generally the appliance course of making
the database and message dealer calls. The applying in most trendy
situations is a stateless microservice which is working in a containerized
setting. It isn’t actually an acceptable place to place the duty
of the coordinator. The coordinator wants to take care of state and recuperate
shortly from failures to commit or rollback, which is tough to
implement on this case.

That is the explanation that whereas XA transactions appear so engaging, they
typically run into points
in apply and are averted. Within the microservices
world, patterns like [transactional-outbox] are most popular over
XA transactions.

Then again most distributed storage techniques implement
two-phase commit throughout a set of partitions, and it really works properly in apply.