Request additional data, buffer expansions, improved logging KIKIMR-14714

ref:c4512da3e927ef4f57f6f5db6e4a0c21c2e6bbf4

5 files changed, 153 insertions, 80 deletions

diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
index e26c0bd845c..5790fa91f8b 100644
--- a/ydb/core/protos/flat_scheme_op.proto
+++ b/ydb/core/protos/flat_scheme_op.proto
@@ -869,6 +869,7 @@ message TS3Settings {
     message TLimits {
         optional uint32 ReadBatchSize = 1 [default = 8388608]; // 8 MB
         optional uint32 WriteBatchSize = 2 [default = 67108864]; // 64 MB
+        optional uint32 ReadBufferSizeLimit = 3 [default = 67108864]; // 64 MB
     };
 
     optional TLimits Limits = 100;
diff --git a/ydb/core/tx/datashard/import_s3.cpp b/ydb/core/tx/datashard/import_s3.cpp
index ecabfe7bb8a..3d4f20c2e22 100644
--- a/ydb/core/tx/datashard/import_s3.cpp
+++ b/ydb/core/tx/datashard/import_s3.cpp
@@ -39,8 +39,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
         }
 
     public:
-        explicit TReadController(ui32 rangeSize)
+        explicit TReadController(ui32 rangeSize, ui64 bufferSizeLimit)
             : RangeSize(rangeSize)
+            , BufferSizeLimit(bufferSizeLimit)
         {
         }
 
@@ -56,6 +57,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
         bool Feed(TString&& portion, TStringBuf& buf) {
             if (Buffer && Pos) {
                 Buffer.remove(0, Pos);
+                Pos = 0;
             }
 
             if (!Buffer) {
@@ -64,19 +66,36 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
                 Buffer.append(portion);
             }
 
-            Pos = Buffer.rfind('\n');
-            if (TString::npos == Pos) {
+            const ui64 pos = Buffer.rfind('\n');
+            if (TString::npos == pos) {
                 return false;
             }
 
-            Pos += 1; // for '\n'
+            Pos = pos + 1; // for '\n'
             buf = TStringBuf(Buffer.data(), Pos);
 
             return true;
         }
 
+        bool CanExpandBuffer(ui64 contentLength, ui64 processedBytes, TString& reason) const {
+            const auto size = Buffer.size();
+
+            if (SumWithSaturation(processedBytes, size) >= contentLength) {
+                reason = "reached end of file";
+                return false;
+            }
+
+            if (size >= BufferSizeLimit) {
+                reason = "reached buffer size limit";
+                return false;
+            }
+
+            return true;
+        }
+
     private:
         const ui32 RangeSize;
+        const ui64 BufferSizeLimit;
 
         TString Buffer;
         ui64 Pos = 0;
@@ -126,8 +145,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     };
 
     void AllocateResource() {
-        IMPORT_LOG_D("AllocateResource"
-            << ": self# " << SelfId());
+        IMPORT_LOG_D("AllocateResource");
 
         const auto* appData = AppData();
         Send(MakeResourceBrokerID(), new TEvResourceBroker::TEvSubmitTask(
@@ -142,18 +160,17 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Handle(TEvResourceBroker::TEvResourceAllocated::TPtr& ev) {
-        IMPORT_LOG_D("Handle TEvResourceBroker::TEvResourceAllocated"
-            << ": self# " << SelfId()
-            << ", taskId# " << ev->Get()->TaskId);
+        IMPORT_LOG_I("Handle TEvResourceBroker::TEvResourceAllocated {"
+            << " TaskId: " << ev->Get()->TaskId
+        << " }");
 
         TaskId = ev->Get()->TaskId;
         Restart();
     }
 
     void Restart() {
-        IMPORT_LOG_D("Restart"
-            << ": self# " << SelfId()
-            << ", attempt# " << Attempt);
+        IMPORT_LOG_N("Restart"
+            << ": attempt# " << Attempt);
 
         if (const TActorId client = std::exchange(Client, TActorId())) {
             Send(client, new TEvents::TEvPoisonPill());
@@ -167,8 +184,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
 
     void HeadObject(const TString& key) {
         IMPORT_LOG_D("HeadObject"
-            << ": self# " << SelfId()
-            << ", key# " << key);
+            << ": key# " << key);
 
         auto request = Model::HeadObjectRequest()
             .WithBucket(Settings.Bucket)
@@ -179,8 +195,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
 
     void GetObject(const TString& key, const std::pair<ui64, ui64>& range) {
         IMPORT_LOG_D("GetObject"
-            << ": self# " << SelfId()
-            << ", key# " << key
+            << ": key# " << key
             << ", range# " << range.first << "-" << range.second);
 
         auto request = Model::GetObjectRequest()
@@ -192,12 +207,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Handle(TEvS3Wrapper::TEvHeadObjectResponse::TPtr& ev) {
-        const auto& result = ev->Get()->Result;
-
-        IMPORT_LOG_D("Handle TEvS3Wrapper::TEvHeadObjectResponse"
-            << ": self# " << SelfId()
-            << ", result# " << result);
+        IMPORT_LOG_D("Handle " << ev->Get()->ToString());
 
+        const auto& result = ev->Get()->Result;
         if (!CheckResult(result, TStringBuf("HeadObject"))) {
             return;
         }
@@ -209,12 +221,9 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Handle(TEvDataShard::TEvS3DownloadInfo::TPtr& ev) {
-        const auto& info = ev->Get()->Info;
-
-        IMPORT_LOG_D("Handle TEvDataShard::TEvS3DownloadInfo"
-            << ": self# " << SelfId()
-            << ", info# " << info);
+        IMPORT_LOG_D("Handle " << ev->Get()->ToString());
 
+        const auto& info = ev->Get()->Info;
         if (!info.DataETag) {
             Send(DataShard, new TEvDataShard::TEvStoreS3DownloadInfo(TxId, {
                 ETag, ProcessedBytes, WrittenBytes, WrittenRows
@@ -226,8 +235,10 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void ProcessDownloadInfo(const TS3Download& info, const TStringBuf marker) {
-        Y_VERIFY(info.DataETag);
+        IMPORT_LOG_N("Process download info at '" << marker << "'"
+            << ": info# " << info);
 
+        Y_VERIFY(info.DataETag);
         if (!CheckETag(*info.DataETag, ETag, marker)) {
             return;
         }
@@ -244,13 +255,10 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Handle(TEvS3Wrapper::TEvGetObjectResponse::TPtr& ev) {
+        IMPORT_LOG_D("Handle " << ev->Get()->ToString());
+
         auto& msg = *ev->Get();
         const auto& result = msg.Result;
-
-        IMPORT_LOG_D("Handle TEvS3Wrapper::TEvGetObjectResponse"
-            << ": self# " << SelfId()
-            << ", result# " << result);
-
         const TStringBuf marker = "GetObject";
 
         if (!CheckResult(result, marker)) {
@@ -262,13 +270,21 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
         }
 
         IMPORT_LOG_T("Content size"
-            << ": self# " << SelfId()
-            << ", processed-bytes# " << ProcessedBytes
+            << ": processed-bytes# " << ProcessedBytes
             << ", content-length# " << ContentLength
             << ", body-size# " << msg.Body.size());
 
         if (!Reader.Feed(std::move(msg.Body), Buffer)) {
-            return Finish(false, "Cannot find new line symbol in data");
+            TString reason;
+            if (!Reader.CanExpandBuffer(ContentLength, ProcessedBytes, reason)) {
+                return Finish(false, TStringBuilder() << "Cannot find new line symbol in data"
+                    << ": " << reason);
+            }
+
+            IMPORT_LOG_W("Cannot find new line symbol, request additional data"
+                << ": processed-bytes# " << ProcessedBytes
+                << ", content-length# " << ContentLength);
+            return GetObject(Settings.DataKey, Reader.NextRange(ContentLength, ProcessedBytes));
         }
 
         ProcessedBytes += Buffer.size();
@@ -325,9 +341,8 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     void UploadRows() {
         const auto& record = RequestBuilder.GetRecord();
 
-        IMPORT_LOG_D("Upload rows"
-            << ": self# " << SelfId()
-            << ", count# " << record->RowsSize()
+        IMPORT_LOG_I("Upload rows"
+            << ": count# " << record->RowsSize()
             << ", size# " << record->ByteSizeLong());
 
         Send(DataShard, new TEvDataShard::TEvUnsafeUploadRowsRequest(TxId, record, {
@@ -336,17 +351,12 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Handle(TEvDataShard::TEvUnsafeUploadRowsResponse::TPtr& ev) {
-        const auto& record = ev->Get()->Record;
-        const auto& info = ev->Get()->Info;
-
-        IMPORT_LOG_D("Handle TEvDataShard::TEvUnsafeUploadRowsResponse"
-            << ": self# " << SelfId()
-            << ", record# " << record.ShortDebugString()
-            << ", info# " << info);
+        IMPORT_LOG_D("Handle " << ev->Get()->ToString());
 
+        const auto& record = ev->Get()->Record;
         switch (record.GetStatus()) {
         case NKikimrTxDataShard::TError::OK:
-            return ProcessDownloadInfo(info, TStringBuf("UploadResponse"));
+            return ProcessDownloadInfo(ev->Get()->Info, TStringBuf("UploadResponse"));
 
         case NKikimrTxDataShard::TError::WRONG_SHARD_STATE: // OVERLOADED
             return RetryUpload();
@@ -357,7 +367,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
 
         default:
             return RestartOrFinish(record.GetErrorDescription());
-        };
+        }
     }
 
     template <typename TResult>
@@ -367,8 +377,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
         }
 
         IMPORT_LOG_E("Error at '" << marker << "'"
-            << ": self# " << SelfId()
-            << ", error# " << result);
+            << ": error# " << result);
         RestartOrFinish(result.GetError().GetMessage().c_str());
 
         return false;
@@ -380,8 +389,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
         }
 
         const TString error = TStringBuilder() << "ETag mismatch at '" << marker << "'"
-            << ": self# " << SelfId()
-            << ", expected# " << expected
+            << ": expected# " << expected
             << ", got# " << got;
 
         IMPORT_LOG_E(error);
@@ -459,9 +467,8 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader>, private TS3User {
     }
 
     void Finish(bool success = true, const TString& error = TString()) {
-        IMPORT_LOG_I("Finish"
-            << ": self# " << SelfId()
-            << ", success# " << success
+        IMPORT_LOG_N("Finish"
+            << ": success# " << success
             << ", error# " << error
             << ", writtenBytes# " << WrittenBytes
             << ", writtenRows# " << WrittenRows);
@@ -493,8 +500,8 @@ public:
         return NKikimrServices::TActivity::IMPORT_S3_DOWNLOADER_ACTOR;
     }
 
-    static constexpr TStringBuf LogPrefix() {
-        return "s3";
+    TStringBuf LogPrefix() const {
+        return LogPrefix_;
     }
 
     explicit TS3Downloader(const TActorId& dataShard, ui64 txId, const NKikimrSchemeOp::TRestoreTask& task, const TTableInfo& tableInfo)
@@ -503,15 +510,15 @@ public:
         , Settings(TS3Settings::FromRestoreTask(task))
         , TableInfo(tableInfo)
         , Scheme(task.GetTableDescription())
+        , LogPrefix_(TStringBuilder() << "s3:" << TxId)
         , Retries(task.GetNumberOfRetries())
-        , Reader(task.GetS3Settings().GetLimits().GetReadBatchSize())
+        , Reader(task.GetS3Settings().GetLimits().GetReadBatchSize(), task.GetS3Settings().GetLimits().GetReadBufferSizeLimit())
     {
     }
 
     void Bootstrap() {
         IMPORT_LOG_D("Bootstrap"
-            << ": self# " << SelfId()
-            << ", attempt# " << Attempt);
+            << ": attempt# " << Attempt);
 
         if (!CheckScheme()) {
             return;
@@ -523,7 +530,7 @@ public:
     STATEFN(StateAllocateResource) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvResourceBroker::TEvResourceAllocated, Handle);
-            cFunc(TEvents::TEvPoisonPill::EventType, NotifyDied);
+            sFunc(TEvents::TEvPoisonPill, NotifyDied);
         }
     }
 
@@ -546,6 +553,7 @@ private:
     const TS3Settings Settings;
     const TTableInfo TableInfo;
     const NKikimrSchemeOp::TTableDescription Scheme;
+    const TString LogPrefix_;
 
     const ui32 Retries;
     ui32 Attempt = 0;
diff --git a/ydb/core/tx/schemeshard/ut_restore.cpp b/ydb/core/tx/schemeshard/ut_restore.cpp
index 5573748eb88..9a84fc19caa 100644
--- a/ydb/core/tx/schemeshard/ut_restore.cpp
+++ b/ydb/core/tx/schemeshard/ut_restore.cpp
@@ -326,6 +326,23 @@ Y_UNIT_TEST_SUITE(TRestoreTests) {
         NKqp::CompareYson(data.YsonStr, content);
     }
 
+    Y_UNIT_TEST(ShouldExpandBuffer) {
+        TTestBasicRuntime runtime;
+
+        const auto data = GenerateTestData("a", 2);
+        const ui32 batchSize = 1;
+
+        Restore(runtime, R"(
+            Name: "Table"
+            Columns { Name: "key" Type: "Utf8" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )", {data}, batchSize);
+
+        auto content = ReadTable(runtime, TTestTxConfig::FakeHiveTablets);
+        NKqp::CompareYson(data.YsonStr, content);
+    }
+
     Y_UNIT_TEST(ShouldSucceedOnSupportedDatatypes) {
         TTestBasicRuntime runtime;
 
@@ -596,7 +613,7 @@ Y_UNIT_TEST_SUITE(TRestoreTests) {
         NKqp::CompareYson(data.YsonStr, content);
     }
 
-    Y_UNIT_TEST(ShouldFailOnFileWithoutNewLines) {
+    void ShouldFailOnFileWithoutNewLines(ui32 batchSize) {
         TTestBasicRuntime runtime;
 
         const auto data = TTestData("\"a1\",\"value1\"", EmptyYsonStr);
@@ -606,12 +623,20 @@ Y_UNIT_TEST_SUITE(TRestoreTests) {
             Columns { Name: "key" Type: "Utf8" }
             Columns { Name: "value" Type: "Utf8" }
             KeyColumnNames: ["key"]
-        )", {data});
+        )", {data}, batchSize);
 
         auto content = ReadTable(runtime, TTestTxConfig::FakeHiveTablets);
         NKqp::CompareYson(data.YsonStr, content);
     }
 
+    Y_UNIT_TEST(ShouldFailOnFileWithoutNewLinesStandardBatch) {
+        ShouldFailOnFileWithoutNewLines(128);
+    }
+
+    Y_UNIT_TEST(ShouldFailOnFileWithoutNewLinesSmallBatch) {
+        ShouldFailOnFileWithoutNewLines(1);
+    }
+
     Y_UNIT_TEST(ShouldFailOnEmptyToken) {
         TTestBasicRuntime runtime;
 
diff --git a/ydb/core/wrappers/s3_wrapper.cpp b/ydb/core/wrappers/s3_wrapper.cpp
index 5d5a21cc033..e3e81b813ff 100644
--- a/ydb/core/wrappers/s3_wrapper.cpp
+++ b/ydb/core/wrappers/s3_wrapper.cpp
@@ -95,6 +95,10 @@ TS3User::~TS3User() {
 class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
     template <typename TEvRequest, typename TEvResponse>
     class TContextBase: public AsyncCallerContext {
+    protected:
+        using TRequest = typename TEvRequest::TRequest;
+        using TOutcome = typename TEvResponse::TOutcome;
+
     public:
         explicit TContextBase(const TActorSystem* sys, const TActorId& sender)
             : AsyncCallerContext()
@@ -108,16 +112,14 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
             return ActorSystem;
         }
 
-        virtual const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) {
+        virtual const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) {
             return ev->Get()->Request;
         }
 
-        void Reply(const typename TEvRequest::TRequest& request,
-                   const typename TEvResponse::TOutcome& outcome) const
-        {
+        void Reply(const TRequest& request, const TOutcome& outcome) const {
             Y_VERIFY(!std::exchange(Replied, true), "Double-reply");
 
-            std::optional<TString> key;
+            typename TEvResponse::TKey key;
             if (request.KeyHasBeenSet()) {
                 key = request.GetKey();
             }
@@ -134,9 +136,7 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
             Send(Sender, ev);
         }
 
-        virtual THolder<IEventBase> MakeResponse(const std::optional<TString>& key,
-                                                 const typename TEvResponse::TOutcome& outcome) const
-        {
+        virtual THolder<IEventBase> MakeResponse(const typename TEvResponse::TKey& key, const TOutcome& outcome) const {
             return MakeHolder<TEvResponse>(key, outcome);
         }
 
@@ -150,6 +150,11 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
 
     template <typename TEvRequest, typename TEvResponse>
     class TOutputStreamContext: public TContextBase<TEvRequest, TEvResponse> {
+    protected:
+        using TRequest = typename TEvRequest::TRequest;
+        using TOutcome = typename TEvResponse::TOutcome;
+
+    private:
         class TOutputStreamBuf: public PreallocatedStreamBuf {
             TOutputStreamBuf(char* data, size_t size)
                 : PreallocatedStreamBuf(reinterpret_cast<unsigned char*>(data), size)
@@ -186,7 +191,7 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
     public:
         using TContextBase<TEvRequest, TEvResponse>::TContextBase;
 
-        const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override {
+        const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override {
             auto& request = ev->Get()->Request;
 
             std::pair<ui64, ui64> range;
@@ -202,9 +207,7 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
         }
 
     protected:
-        THolder<IEventBase> MakeResponse(const std::optional<TString>& key,
-                                         const typename TEvResponse::TOutcome& outcome) const override
-        {
+        THolder<IEventBase> MakeResponse(const typename TEvResponse::TKey& key, const TOutcome& outcome) const override {
             if (outcome.IsSuccess()) {
                 return MakeHolder<TEvResponse>(key, outcome, std::move(Buffer));
             } else {
@@ -219,6 +222,11 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
 
     template <typename TEvRequest, typename TEvResponse>
     class TInputStreamContext: public TContextBase<TEvRequest, TEvResponse> {
+    protected:
+        using TRequest = typename TEvRequest::TRequest;
+        using TOutcome = typename TEvResponse::TOutcome;
+
+    private:
         class TInputStreamBuf: public PreallocatedStreamBuf {
             TInputStreamBuf(char* data, size_t size)
                 : PreallocatedStreamBuf(reinterpret_cast<unsigned char*>(data), size)
@@ -240,7 +248,7 @@ class TS3Wrapper: public TActor<TS3Wrapper>, private TS3User {
     public:
         using TContextBase<TEvRequest, TEvResponse>::TContextBase;
 
-        const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override {
+        const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override {
             auto& request = ev->Get()->Request;
 
             Buffer = std::move(ev->Get()->Body);
diff --git a/ydb/core/wrappers/s3_wrapper.h b/ydb/core/wrappers/s3_wrapper.h
index c623113e9a0..369621f29e2 100644
--- a/ydb/core/wrappers/s3_wrapper.h
+++ b/ydb/core/wrappers/s3_wrapper.h
@@ -16,6 +16,8 @@
 
 #include <ydb/core/base/events.h>
 
+#include <util/string/builder.h>
+
 namespace NKikimr {
 namespace NWrappers {
 
@@ -35,6 +37,12 @@ struct TEvS3Wrapper {
         {
         }
 
+        TString ToString() const override {
+            return TStringBuilder() << this->ToStringHeader() << " {"
+                << " Request: " << Request
+            << " }";
+        }
+
         using TBase = TGenericRequest<TDerived, EventType, TRequest>;
     };
 
@@ -50,6 +58,13 @@ struct TEvS3Wrapper {
         {
         }
 
+        TString ToString() const override {
+            return TStringBuilder() << this->ToStringHeader() << " {"
+                << " Request: " << this->Request
+                << " Body: " << Body.size() << "b"
+            << " }";
+        }
+
         using TBase = TRequestWithBody<TDerived, EventType, T>;
     };
 
@@ -57,11 +72,12 @@ struct TEvS3Wrapper {
     struct TGenericResponse: public TEventLocal<TDerived, EventType> {
         using TOutcome = Aws::Utils::Outcome<T, Aws::S3::S3Error>;
         using TResult = Aws::Utils::Outcome<U, Aws::S3::S3Error>;
+        using TKey = std::optional<TString>;
 
-        std::optional<TString> Key;
+        TKey Key;
         TResult Result;
 
-        explicit TGenericResponse(const std::optional<TString>& key, const TOutcome& outcome)
+        explicit TGenericResponse(const TKey& key, const TOutcome& outcome)
             : Key(key)
             , Result(TDerived::ResultFromOutcome(outcome))
         {
@@ -71,27 +87,42 @@ struct TEvS3Wrapper {
             return outcome;
         }
 
+        TString ToString() const override {
+            return TStringBuilder() << this->ToStringHeader() << " {"
+                << " Key: " << (Key ? "null" : *Key)
+                << " Result: " << Result
+            << " }";
+        }
+
         using TBase = TGenericResponse<TDerived, EventType, T, U>;
     };
 
     template <typename TDerived, ui32 EventType, typename T, typename U>
     struct TResponseWithBody: public TGenericResponse<TDerived, EventType, T, U> {
         using TGeneric = TGenericResponse<TDerived, EventType, T, U>;
+        using TKey = typename TGeneric::TKey;
 
         TString Body;
 
-        explicit TResponseWithBody(const std::optional<TString>& key, const typename TGeneric::TOutcome& outcome)
+        explicit TResponseWithBody(const TKey& key, const typename TGeneric::TOutcome& outcome)
             : TGeneric(key, outcome)
         {
         }
 
-        explicit TResponseWithBody(const std::optional<TString>& key, const typename TGeneric::TOutcome& outcome,
-                                   TString&& body)
+        explicit TResponseWithBody(const TKey& key, const typename TGeneric::TOutcome& outcome, TString&& body)
             : TGeneric(key, outcome)
             , Body(std::move(body))
         {
         }
 
+        TString ToString() const override {
+            return TStringBuilder() << this->ToStringHeader() << " {"
+                << " Key: " << (this->Key ? "null" : *this->Key)
+                << " Result: " << this->Result
+                << " Body: " << Body.size() << "b"
+            << " }";
+        }
+
         using TBase = TResponseWithBody<TDerived, EventType, T, U>;
     };