feat(native): Add http2 support for HTTP client (#26439)

```
== RELEASE NOTES ==

General Changes
* Add http2 support for HTTP client

```

Author: kewang1024

presto-docs/src/main/sphinx/presto_cpp/properties.rst

@@ -561,6 +561,17 @@ Exchange Properties
 
   Maximum wait time for exchange request in seconds.
 
+HTTP Client Properties
+----------------------
+
+``http-client.http2-enabled``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+Specifies whether HTTP/2 should be enabled for HTTP client.
+
 Memory Checker Properties
 -------------------------
 

presto-native-execution/presto_cpp/main/PrestoExchangeSource.cpp

@@ -278,9 +278,11 @@ void PrestoExchangeSource::processDataResponse(
     return;
   }
   auto* headers = response->headers();
-  VELOX_CHECK(
-      !headers->getIsChunked(),
-      "Chunked http transferring encoding is not supported.");
+  if (!SystemConfig::instance()->httpClientHttp2Enabled()) {
+    VELOX_CHECK(
+        !headers->getIsChunked(),
+        "Chunked http transferring encoding is not supported.");
+  }
   const uint64_t contentLength =
       atol(headers->getHeaders()
                .getSingleOrEmpty(proxygen::HTTP_HEADER_CONTENT_LENGTH)

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -259,8 +259,10 @@ void PrestoServer::run() {
             "Https Client Certificates are not configured correctly");
       }
 
-      sslContext_ =
-          util::createSSLContext(optionalClientCertPath.value(), ciphers);
+      sslContext_ = util::createSSLContext(
+          optionalClientCertPath.value(),
+          ciphers,
+          systemConfig->httpClientHttp2Enabled());
     }
 
     if (systemConfig->internalCommunicationJwtEnabled()) {

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -232,6 +232,7 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kLogNumZombieTasks, 20),
           NUM_PROP(kAnnouncementMaxFrequencyMs, 30'000), // 30s
           NUM_PROP(kHeartbeatFrequencyMs, 0),
+          BOOL_PROP(kHttpClientHttp2Enabled, false),
           STR_PROP(kExchangeMaxErrorDuration, "3m"),
           STR_PROP(kExchangeRequestTimeout, "20s"),
           STR_PROP(kExchangeConnectTimeout, "20s"),
@@ -850,6 +851,10 @@ uint64_t SystemConfig::heartbeatFrequencyMs() const {
   return optionalProperty<uint64_t>(kHeartbeatFrequencyMs).value();
 }
 
+bool SystemConfig::httpClientHttp2Enabled() const {
+  return optionalProperty<bool>(kHttpClientHttp2Enabled).value();
+}
+
 std::chrono::duration<double> SystemConfig::exchangeMaxErrorDuration() const {
   return velox::config::toDuration(
       optionalProperty(kExchangeMaxErrorDuration).value());

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -648,6 +648,10 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kHeartbeatFrequencyMs{
       "heartbeat-frequency-ms"};
 
+  /// Whether HTTP/2 is enabled for HTTP client connections.
+  static constexpr std::string_view kHttpClientHttp2Enabled{
+      "http-client.http2-enabled"};
+
   static constexpr std::string_view kExchangeMaxErrorDuration{
       "exchange.max-error-duration"};
 
@@ -1029,6 +1033,8 @@ class SystemConfig : public ConfigBase {
 
   uint64_t heartbeatFrequencyMs() const;
 
+  bool httpClientHttp2Enabled() const;
+
   std::chrono::duration<double> exchangeMaxErrorDuration() const;
 
   std::chrono::duration<double> exchangeRequestTimeoutMs() const;

presto-native-execution/presto_cpp/main/common/Utils.cpp

@@ -31,13 +31,18 @@ DateTime toISOTimestamp(uint64_t timeMilli) {
 
 std::shared_ptr<folly::SSLContext> createSSLContext(
     const std::string& clientCertAndKeyPath,
-    const std::string& ciphers) {
+    const std::string& ciphers,
+    bool http2Enabled) {
   try {
     auto sslContext = std::make_shared<folly::SSLContext>();
     sslContext->loadCertKeyPairFromFiles(
         clientCertAndKeyPath.c_str(), clientCertAndKeyPath.c_str());
     sslContext->setCiphersOrThrow(ciphers);
-    sslContext->setAdvertisedNextProtocols({"http/1.1"});
+    if (http2Enabled) {
+      sslContext->setAdvertisedNextProtocols({"h2", "http/1.1"});
+    } else {
+      sslContext->setAdvertisedNextProtocols({"http/1.1"});
+    }
     return sslContext;
   } catch (const std::exception& ex) {
     LOG(FATAL) << fmt::format(

presto-native-execution/presto_cpp/main/common/Utils.h

@@ -30,7 +30,8 @@ DateTime toISOTimestamp(uint64_t timeMilli);
 
 std::shared_ptr<folly::SSLContext> createSSLContext(
     const std::string& clientCertAndKeyPath,
-    const std::string& ciphers);
+    const std::string& ciphers,
+    bool http2Enabled);
 
 /// Returns current process-wide CPU time in nanoseconds.
 long getProcessCpuTimeNs();

presto-native-execution/presto_cpp/main/http/HttpClient.cpp

@@ -18,6 +18,7 @@
 #endif // PRESTO_ENABLE_JWT
 #include <folly/io/async/EventBaseManager.h>
 #include <folly/synchronization/Latch.h>
+#include <proxygen/lib/http/codec/CodecProtocol.h>
 #include <velox/common/base/Exceptions.h>
 #include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/main/common/Counters.h"
@@ -201,13 +202,22 @@ class ResponseHandler : public proxygen::HTTPTransactionHandler {
     return promise_.getSemiFuture();
   }
 
-  void setTransaction(proxygen::HTTPTransaction* /* txn */) noexcept override {}
+  void setTransaction(proxygen::HTTPTransaction* txn) noexcept override {
+    if (txn) {
+      protocol_ = txn->getTransport().getCodec().getProtocol();
+    }
+  }
+
   void detachTransaction() noexcept override {
     self_.reset();
   }
 
   void onHeadersComplete(
       std::unique_ptr<proxygen::HTTPMessage> msg) noexcept override {
+    if (protocol_.has_value()) {
+      VLOG(2) << "HttpClient received response of "
+              << proxygen::getCodecProtocolString(protocol_.value());
+    }
     response_ = std::make_unique<HttpResponse>(
         std::move(msg),
         client_->memoryPool(),
@@ -268,6 +278,7 @@ class ResponseHandler : public proxygen::HTTPTransactionHandler {
   folly::Promise<std::unique_ptr<HttpResponse>> promise_;
   std::shared_ptr<ResponseHandler> self_;
   std::shared_ptr<HttpClient> client_;
+  std::optional<proxygen::CodecProtocol> protocol_;
 };
 
 // Responsible for making an HTTP request. The request will be made in 2

presto-native-execution/presto_cpp/main/http/HttpClient.h

@@ -214,9 +214,7 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
 
 class RequestBuilder {
  public:
-  RequestBuilder() {
-    headers_.setHTTPVersion(1, 1);
-  }
+  RequestBuilder() {}
 
   RequestBuilder& method(proxygen::HTTPMethod method) {
     headers_.setMethod(method);

presto-native-execution/presto_cpp/main/tests/AnnouncerTest.cpp

@@ -104,7 +104,7 @@ class AnnouncerTestSuite : public ::testing::TestWithParam<bool> {
 
     std::string keyPath = getCertsPath("client_ca.pem");
     std::string ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384,AES256-GCM-SHA384";
-    sslContext_ = util::createSSLContext(keyPath, ciphers);
+    sslContext_ = util::createSSLContext(keyPath, ciphers, false);
   }
 
  protected: