Merge pull request ClickHouse#86758 from ClickHouse/backport/25.8/86631

Backport ClickHouse#86631 to 25.8: Fixing keep-alive on inter-server calls.

Author: clickhouse-gh[bot]

src/Interpreters/InterserverIOHandler.h

@@ -3,6 +3,7 @@
 #include <Common/ActionBlocker.h>
 #include <Common/Exception.h>
 #include <Common/SharedMutex.h>
+#include <IO/ReadBuffer.h>
 #include <base/types.h>
 
 #include <map>
@@ -34,7 +35,7 @@ class InterserverIOEndpoint
 {
 public:
     virtual std::string getId(const std::string & path) const = 0;
-    virtual void processQuery(const HTMLForm & params, ReadBuffer & body, WriteBuffer & out, HTTPServerResponse & response) = 0;
+    virtual void processQuery(const HTMLForm & params, ReadBufferPtr body, WriteBuffer & out, HTTPServerResponse & response) = 0;
     virtual ~InterserverIOEndpoint() = default;
 
     /// You need to stop the data transfer if blocker is activated.

src/Server/InterserverIOHTTPHandler.cpp

@@ -61,8 +61,6 @@ void InterserverIOHTTPHandler::processQuery(HTTPServerRequest & request, HTTPSer
     String endpoint_name = params.get("endpoint");
     bool compress = params.get("compress") == "true";
 
-    auto input_stream_with_body = request.getStream();
-
     auto endpoint = server.context()->getInterserverIOHandler().getEndpoint(endpoint_name);
     /// Locked for read while query processing
     std::shared_lock lock(endpoint->rwlock);
@@ -72,13 +70,15 @@ void InterserverIOHTTPHandler::processQuery(HTTPServerRequest & request, HTTPSer
     if (compress)
     {
         CompressedWriteBuffer compressed_out(*output);
-        endpoint->processQuery(params, *input_stream_with_body, compressed_out, response);
+        endpoint->processQuery(params, request.getStream(), compressed_out, response);
         compressed_out.finalize();
     }
     else
     {
-        endpoint->processQuery(params, *input_stream_with_body, *output, response);
+        endpoint->processQuery(params, request.getStream(), *output, response);
     }
+    /// Make sure that request stream is not used after this function.
+    assert(request.getStream().use_count() == 2);
 }
 
 

src/Storages/MergeTree/DataPartsExchange.cpp

@@ -123,8 +123,11 @@ std::string Service::getId(const std::string & node_id) const
     return getEndpointId(node_id);
 }
 
-void Service::processQuery(const HTMLForm & params, ReadBuffer & /*body*/, WriteBuffer & out, HTTPServerResponse & response)
+void Service::processQuery(const HTMLForm & params, ReadBufferPtr body, WriteBuffer & out, HTTPServerResponse & response)
 {
+    // nothing to read from body
+    body.reset();
+
     int client_protocol_version = parse<int>(params.get("client_protocol_version", "0"));
 
     String part_name = params.get("part");

src/Storages/MergeTree/DataPartsExchange.h

@@ -9,6 +9,7 @@
 #include <IO/ConnectionTimeouts.h>
 #include <Common/Throttler.h>
 #include <Common/ActionBlocker.h>
+#include <IO/ReadBuffer.h>
 
 
 namespace zkutil
@@ -37,7 +38,7 @@ class Service final : public InterserverIOEndpoint
     Service & operator=(const Service &) = delete;
 
     std::string getId(const std::string & node_id) const override;
-    void processQuery(const HTMLForm & params, ReadBuffer & body, WriteBuffer & out, HTTPServerResponse & response) override;
+    void processQuery(const HTMLForm & params, ReadBufferPtr body, WriteBuffer & out, HTTPServerResponse & response) override;
 
 private:
     MergeTreeData::DataPartPtr findPart(const String & name);