YT-27827: TPingerTransaction use BlockingGet instead of WaitFor in destructor

commit_hash:90bbe36635e0d48c81c153567dcedf28f103efbe

1 files changed, 72 insertions, 55 deletions

diff --git a/yt/cpp/mapreduce/http_client/raw_client.cpp b/yt/cpp/mapreduce/http_client/raw_client.cpp
index bf63a7ae3bb..8ef2a349a3e 100644
--- a/yt/cpp/mapreduce/http_client/raw_client.cpp
+++ b/yt/cpp/mapreduce/http_client/raw_client.cpp
@@ -65,6 +65,17 @@ void CheckError(const TString& requestId, NHttp::IResponsePtr response)
     }
 }
 
+void SetMutationId(TNode& params, TMutationId& mutationId)
+{
+    if (mutationId.IsEmpty()) {
+        params["retry"] = false;
+        mutationId = GenerateMutationId();
+    } else {
+        params["retry"] = true;
+    }
+    params["mutation_id"] = GetGuidAsString(mutationId);
+}
+
 } // anonymous namespace
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -308,68 +319,20 @@ TTransactionId THttpRawClient::StartTransaction(
 
 void THttpRawClient::PingTransaction(const TTransactionId& transactionId)
 {
-    auto traceContext = Context_.Config->EnableClientTracing
-        ? NTracing::CreateTraceContextFromCurrent("ping_tx")
-        : nullptr;
-    NTracing::TCurrentTraceContextGuard traceContextGuard(traceContext);
-
-    std::call_once(PingClientInitOnceFlag_, [this] () {
-        InitPingClient();
-    });
-
-    auto url = TString::Join(Context_.UseTLS ? "https://" : "http://", Context_.ServerName, "/api/", Context_.Config->ApiVersion, "/ping_tx");
-    auto headers = New<NHttp::THeaders>();
-    auto requestId = CreateGuidAsString();
-
-    headers->Add("Host", url);
-    headers->Add("User-Agent", TProcessState::Get()->ClientVersion);
-
-    if (const auto& serviceTicketAuth = Context_.ServiceTicketAuth) {
-        const auto serviceTicket = serviceTicketAuth->Ptr->IssueServiceTicket();
-        headers->Add("X-Ya-Service-Ticket", serviceTicket);
-    } else if (const auto& token = Context_.Token; !token.empty()) {
-        headers->Add("Authorization", "OAuth " + token);
-    }
-
-    headers->Add("Transfer-Encoding", "chunked");
-    headers->Add("X-YT-Correlation-Id", requestId);
-    headers->Add("X-YT-Header-Format", "<format=text>yson");
-    headers->Add("Content-Encoding", "identity");
-    headers->Add("Accept-Encoding", "identity");
-
-    if (traceContext) {
-        auto traceparent = FormatTraceParentHeader(traceContext->GetTraceId(), traceContext->GetSpanId());
-        headers->Add("traceparent", traceparent);
-    }
-
     TNode node;
     node["transaction_id"] = GetGuidAsString(transactionId);
     auto strParams = NodeToYsonString(node);
 
-    YT_LOG_DEBUG("REQ %v - sending request (HostName: %v; Method POST %v; X-YT-Parameters (sent in body): %v)",
-        requestId,
-        Context_.ServerName,
-        url,
-        strParams);
-
-    auto response = NConcurrency::WaitFor(PingHttpClient_->Post(url, TSharedRef::FromString(strParams), headers))
-        .ValueOrThrow();
-    CheckError(requestId, response);
-
-    YT_LOG_DEBUG("RSP %v - received response %v bytes. (%v)",
-        requestId,
-        response->ReadAll().size(),
-        strParams);
+    PostAsync("ping_tx", node);
 }
 
 void THttpRawClient::AbortTransaction(
     TMutationId& mutationId,
     const TTransactionId& transactionId)
 {
-    THttpHeader header("POST", "abort_tx");
-    header.AddMutationId();
-    header.MergeParameters(NRawClient::SerializeParamsForAbortTransaction(transactionId));
-    RequestWithoutRetry(Context_, mutationId, header)->GetResponse();
+    TNode params = NRawClient::SerializeParamsForAbortTransaction(transactionId);
+    SetMutationId(params, mutationId);
+    PostAsync("abort_tx", params);
 }
 
 void THttpRawClient::CommitTransaction(
@@ -1257,7 +1220,7 @@ IRawClientPtr THttpRawClient::Clone(const TClientContext& context)
     return ::MakeIntrusive<THttpRawClient>(context);
 }
 
-void THttpRawClient::InitPingClient() {
+void THttpRawClient::InitAsyncClient() {
     auto httpPoller = NConcurrency::CreateThreadPoolPoller(
         Context_.Config->AsyncHttpClientThreads,
         "tx_http_client_poller");
@@ -1265,14 +1228,68 @@ void THttpRawClient::InitPingClient() {
     if (Context_.UseTLS) {
         auto httpsClientConfig = NYT::New<NHttps::TClientConfig>();
         httpsClientConfig->MaxIdleConnections = 16;
-        PingHttpClient_ = NHttps::CreateClient(std::move(httpsClientConfig), std::move(httpPoller));
+        AsyncHttpClient_ = NHttps::CreateClient(std::move(httpsClientConfig), std::move(httpPoller));
     } else {
         auto httpClientConfig = NYT::New<NHttp::TClientConfig>();
         httpClientConfig->MaxIdleConnections = 16;
-        PingHttpClient_ = NHttp::CreateClient(std::move(httpClientConfig), std::move(httpPoller));
+        AsyncHttpClient_ = NHttp::CreateClient(std::move(httpClientConfig), std::move(httpPoller));
     }
 }
 
+void THttpRawClient::PostAsync(const TString& command, TNode params)
+{
+    auto traceContext = Context_.Config->EnableClientTracing
+        ? NTracing::CreateTraceContextFromCurrent("ping_tx")
+        : nullptr;
+    NTracing::TCurrentTraceContextGuard traceContextGuard(traceContext);
+
+    std::call_once(AsyncHttpClientInitOnceFlag_, [this] () {
+        InitAsyncClient();
+    });
+
+    auto url = TString::Join(Context_.UseTLS ? "https://" : "http://", Context_.ServerName, "/api/", Context_.Config->ApiVersion, "/", command);
+    auto headers = New<NHttp::THeaders>();
+    auto requestId = CreateGuidAsString();
+
+    headers->Add("Host", url);
+    headers->Add("User-Agent", TProcessState::Get()->ClientVersion);
+
+    if (const auto& serviceTicketAuth = Context_.ServiceTicketAuth) {
+        const auto serviceTicket = serviceTicketAuth->Ptr->IssueServiceTicket();
+        headers->Add("X-Ya-Service-Ticket", serviceTicket);
+    } else if (const auto& token = Context_.Token; !token.empty()) {
+        headers->Add("Authorization", "OAuth " + token);
+    }
+
+    headers->Add("Transfer-Encoding", "chunked");
+    headers->Add("X-YT-Correlation-Id", requestId);
+    headers->Add("X-YT-Header-Format", "<format=text>yson");
+    headers->Add("Content-Encoding", "identity");
+    headers->Add("Accept-Encoding", "identity");
+
+    if (traceContext) {
+        auto traceparent = FormatTraceParentHeader(traceContext->GetTraceId(), traceContext->GetSpanId());
+        headers->Add("traceparent", traceparent);
+    }
+
+    auto strParams = NodeToYsonString(params);
+
+    YT_LOG_DEBUG("REQ %v - sending request (HostName: %v; Method POST %v; X-YT-Parameters (sent in body): %v)",
+        requestId,
+        Context_.ServerName,
+        url,
+        strParams);
+
+    auto response = NConcurrency::WaitFor(AsyncHttpClient_->Post(url, TSharedRef::FromString(strParams), headers))
+        .ValueOrThrow();
+    CheckError(requestId, response);
+
+    YT_LOG_DEBUG("RSP %v - received response %v bytes. (%v)",
+        requestId,
+        response->ReadAll().size(),
+        strParams);
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NDetail