[yugabyte#13659] DocDB: Fix Prepare for the failed transaction

Summary:
YBTransaction starts with looking up for status tablet.
When this lookup is failed the transaction is marked as failed.

We lack failure state check at Prepare, so calling Prepare after transaction failure will result in adding callback to waiters list.
But no more action will be taken, so callback will never be notified.

The side effect is master leader restarts can cause prepared statement connections to hang.

This diff fixed the issue by checking failure status in Prepare.

Test Plan:
Launch AWS cluster.
Run the following workload against it:
java  -jar yb-sample-apps.jar --workload SqlDataLoad  --num_writes -1 --num_threads_write 5 --num_threads_read 0 --num_reads 0 --num_unique_keys 100000000000 --batch_size 774 --num_value_columns 8 --nodes $NODES
Wait 10 minutes.
Stop node, that runs master leader, using AWS console.

W/o the fix there will be stuck queries.

Reviewers: esheng, rthallam

Reviewed By: esheng, rthallam

Subscribers: rthallam, ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D18994

build-support/post_install.sh

@@ -152,7 +152,11 @@ if [[ $install_mode == "true" ]]; then
    exit 1
   fi
 
-  ln -sfT "$linuxbrew_dir" "$BREW_HOME"
+  if [[ "$linuxbrew_dir" != "$BREW_HOME" ]]; then
+    ln -sfT "$linuxbrew_dir" "$BREW_HOME"
+  else
+    echo "Skipping linuxbrew symlink, since it already has necessary length: $linuxbrew_dir"
+  fi
 
   # We are relying on the fact that $distribution_dir is not a symlink. We don't want to add symlink
   # resolution to the find command because someone may accidentally add a symlink pointing to a

java/yb-cql/src/test/java/org/yb/cql/TestTransaction.java

@@ -624,11 +624,11 @@ public void testTimeout() throws Exception {
       LOG.info("Initial expired transactions = {}", initialExpiredTransactions);
 
       try {
-        thrown.expect(com.datastax.driver.core.exceptions.OperationTimedOutException.class);
+        thrown.expect(com.datastax.driver.core.exceptions.DriverException.class);
         session.execute("begin transaction" +
                         "  insert into test_timeout (k, v) values (1, 1);" +
                         "end transaction;");
-      } catch (com.datastax.driver.core.exceptions.OperationTimedOutException e) {
+      } catch (com.datastax.driver.core.exceptions.DriverException e) {
         int currentExpiredTransactions = getExpiredTransactionsCount();
         LOG.info("Current expired transactions = {}", currentExpiredTransactions);
         assertTrue(currentExpiredTransactions > initialExpiredTransactions);

src/yb/client/client.cc

@@ -488,6 +488,7 @@ Status YBClientBuilder::DoBuild(rpc::Messenger* messenger, std::unique_ptr<YBCli
   // is clearer to see "mastercb" instead.
   ThreadPoolBuilder tpb(data_->client_name_ + "cb");
   tpb.set_max_threads(narrow_cast<int>(callback_threadpool_size));
+  tpb.set_min_threads(1);
   std::unique_ptr<ThreadPool> tp;
   RETURN_NOT_OK_PREPEND(
       tpb.Build(&tp),

src/yb/client/transaction.cc

@@ -321,6 +321,16 @@ class YBTransaction::Impl final : public internal::TxnBatcherIf {
       }
       const bool defer = !ready_ || *promotion_started;
 
+      if (!status_.ok()) {
+        auto status = status_;
+        lock.unlock();
+        VLOG_WITH_PREFIX(2) << "Prepare, transaction already failed: " << status;
+        if (waiter) {
+          waiter(status);
+        }
+        return false;
+      }
+
       if (!defer || initial) {
         PrepareOpsGroups(initial, ops_info->groups);
       }
@@ -802,11 +812,6 @@ class YBTransaction::Impl final : public internal::TxnBatcherIf {
     return metadata_;
   }
 
-  void StartHeartbeat() {
-    VLOG_WITH_PREFIX(2) << __PRETTY_FUNCTION__;
-    RequestStatusTablet(TransactionRpcDeadline());
-  }
-
   void SetActiveSubTransaction(SubTransactionId id) {
     VLOG_WITH_PREFIX(4) << "set active sub txn=" << id
                         << ", subtransaction_=" << subtransaction_.ToString();
@@ -2124,13 +2129,6 @@ Trace* YBTransaction::trace() {
   return impl_->trace();
 }
 
-YBTransactionPtr YBTransaction::Take(
-    TransactionManager* manager, const TransactionMetadata& metadata) {
-  auto result = std::make_shared<YBTransaction>(manager, metadata, PrivateOnlyTag());
-  result->impl_->StartHeartbeat();
-  return result;
-}
-
 void YBTransaction::SetActiveSubTransaction(SubTransactionId id) {
   return impl_->SetActiveSubTransaction(id);
 }

src/yb/client/transaction.h

@@ -149,10 +149,6 @@ class YBTransaction : public std::enable_shared_from_this<YBTransaction> {
   // So this transaction could be used by some other application instance.
   Result<TransactionMetadata> Release();
 
-  // Creates transaction by metadata, could be used in pair with release to transfer transaction
-  // between application instances.
-  static YBTransactionPtr Take(TransactionManager* manager, const TransactionMetadata& metadata);
-
   void SetActiveSubTransaction(SubTransactionId id);
 
   Status RollbackToSubTransaction(SubTransactionId id, CoarseTimePoint deadline);