Extract MlpWriter class (#29297)

Author: nshestakov

ydb/core/http_proxy/ut/sqs_topic_ut.cpp

@@ -279,6 +279,7 @@ Y_UNIT_TEST_SUITE(TestSqsTopicHttpProxy) {
                 }}
             });
 
+            Cerr << (TStringBuilder() << "json = " << WriteJson(json, true, true) << '\n');
             UNIT_ASSERT_VALUES_EQUAL(json["Successful"].GetArray().size(), 3);
             auto succesful0 = json["Successful"][0];
             UNIT_ASSERT_VALUES_EQUAL(succesful0["Id"], "Id-0");

ydb/core/persqueue/pqtablet/partition/mlp/mlp_common.cpp

@@ -66,7 +66,7 @@ std::unique_ptr<TEvPQ::TEvSetClientInfo> MakeEvCommit(
         consumer.GetGeneration(),
         0, // step
         TActorId{} // pipeClient
-    );    
+    );
 }
 
 std::unique_ptr<TEvPersQueue::TEvHasDataInfo> MakeEvHasData(

ydb/core/persqueue/pqtablet/partition/mlp/mlp_consumer.cpp

@@ -669,7 +669,7 @@ bool TConsumerActor::FetchMessagesIfNeeded() {
 
 void TConsumerActor::Handle(TEvPQ::TEvProxyResponse::TPtr& ev) {
     LOG_D("Handle TEvPQ::TEvProxyResponse");
-    
+
     AFL_ENSURE(IsSucess(ev))("e", ev->Get()->Response->DebugString());
 }
 
@@ -710,7 +710,7 @@ void TConsumerActor::Handle(TEvPersQueue::TEvResponse::TPtr& ev) {
         AFL_ENSURE(res)("o", result.GetOffset());
 
         for (auto& attr : *proto.MutableMessageMeta()) {
-            if (attr.key() == NMessageConsts::MessageId) {
+            if (attr.key() == MESSAGE_KEY) {
                 messageGroupId = std::move(*attr.mutable_value());
             } else if (attr.key() == NMessageConsts::DelaySeconds) {
                 delaySeconds = std::stoul(attr.value());

ydb/core/persqueue/pqtablet/partition/mlp/mlp_consumer.h

@@ -77,7 +77,7 @@ class TConsumerActor : public TBaseTabletActor<TConsumerActor>
 
     void CommitIfNeeded();
     void UpdateStorageConfig();
-    
+
     size_t RequiredToFetchMessageCount() const;
     void SendToPQTablet(std::unique_ptr<IEventBase> ev);
 

ydb/core/persqueue/pqtablet/partition/mlp/mlp_storage.cpp

@@ -725,7 +725,7 @@ void TStorage::MoveBaseDeadline() {
     }
 
     auto newBaseDeadline = TrimToSeconds(TimeProvider->Now(), false);
-    auto newBaseWriteTimestamp = BaseWriteTimestamp + 
+    auto newBaseWriteTimestamp = BaseWriteTimestamp +
         (SlowMessages.empty() ? TDuration::Seconds(Messages.front().WriteTimestampDelta)
             : TDuration::Seconds(SlowMessages.begin()->second.WriteTimestampDelta));
 
@@ -816,7 +816,7 @@ TString TStorage::DebugString() const {
          << " BaseDeadline: " << BaseDeadline.ToString()
          << " BaseWriteTimestamp: " << BaseWriteTimestamp.ToString()
          << " Messages: [";
-    
+
     auto dump = [&](const auto offset, const auto& message, auto zone) {
         sb << zone <<"{" << offset << ", "
             << static_cast<EMessageStatus>(message.Status) << ", "

ydb/core/persqueue/pqtablet/partition/mlp/mlp_storage_ut.cpp

@@ -77,7 +77,7 @@ struct TUtils {
 
             utils.AssertEquals(*this);
         }
-    } 
+    }
 
     NKikimrPQ::TMLPStorageSnapshot CreateSnapshot() {
         // Clear batch
@@ -1410,7 +1410,7 @@ Y_UNIT_TEST(StorageSerialization_WAL_WithMoveBaseTime_Deadline) {
             UNIT_ASSERT(message);
             UNIT_ASSERT_VALUES_EQUAL(message->DeadlineDelta, 5);
         }
-        
+
         timeProvider->Tick(TDuration::Seconds(3));
         storage.MoveBaseDeadline();
         {

ydb/core/persqueue/public/constants.h

@@ -7,6 +7,8 @@ namespace NKikimr::NPQ {
 static const TString CLIENTID_COMPACTION_CONSUMER = "__ydb_compaction_consumer";
 static const TString CLIENTID_WITHOUT_CONSUMER = "$without_consumer";
 
+static constexpr TStringBuf MESSAGE_KEY = "__key";
+
 static constexpr ui32 METRICS_LEVEL_DISABLED = 0;
 static constexpr ui32 METRICS_LEVEL_DATABASE = 1;
 static constexpr ui32 METRICS_LEVEL_OBJECT = 2;

ydb/core/persqueue/public/mlp/mlp.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ydb/core/persqueue/events/events.h>
+#include <ydb/core/persqueue/public/describer/describer.h>
 #include <ydb/public/api/protos/ydb_status_codes.pb.h>
 #include <ydb/public/api/protos/ydb_topic.pb.h>
 #include <ydb/library/actors/core/actorsystem_fwd.h>
@@ -13,6 +14,7 @@ namespace NKikimr::NPQ::NMLP {
 
 enum EEv : ui32 {
     EvReadResponse = InternalEventSpaceBegin(NPQ::NEvents::EServices::MLP),
+    EvWriteResponse,
     EvChangeResponse,
     EvEnd
 };
@@ -22,6 +24,18 @@ struct TMessageId {
     ui64 Offset;
 };
 
+struct TEvWriteResponse : public NActors::TEventLocal<TEvWriteResponse, EEv::EvWriteResponse> {
+
+    NDescriber::EStatus DescribeStatus;
+
+    struct TMessage {
+        size_t Index;
+        // if message was written successfully, it will be set
+        std::optional<TMessageId> MessageId;
+    };
+    std::vector<TMessage> Messages;
+};
+
 struct TEvReadResponse : public NActors::TEventLocal<TEvReadResponse, EEv::EvReadResponse> {
 
     TEvReadResponse(Ydb::StatusIds::StatusCode status = Ydb::StatusIds::SUCCESS, TString&& errorDescription = {})
@@ -41,7 +55,6 @@ struct TEvReadResponse : public NActors::TEventLocal<TEvReadResponse, EEv::EvRea
         TInstant SentTimestamp;
         TString MessageGroupId;
     };
-    // The original topic path (from request) -> TopicInfo
     std::vector<TMessage> Messages;
 };
 
@@ -61,11 +74,29 @@ struct TEvChangeResponse : public NActors::TEventLocal<TEvChangeResponse, EEv::E
         TMessageId MessageId;
         bool Success = false;
     };
-    // The original topic path (from request) -> TopicInfo
     std::vector<TResult> Messages;
 };
 
+struct TWriterSettings {
+    TString DatabasePath;
+    TString TopicName;
+
+    struct TMessage {
+        size_t Index;
+        TString MessageBody;
+        std::optional<TString> MessageGroupId;
+        std::optional<TString> MessageDeduplicationId;
+        std::optional<TString> SerializedMessageAttributes;
+        TDuration Delay;
+    };
+    std::vector<TMessage> Messages;
+
+    bool ShouldBeCharged = false;
+
+    TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+};
 
+IActor* CreateWriter(const NActors::TActorId& parentId, TWriterSettings&& settings);
 
 struct TReaderSettings {
     TString DatabasePath;

ydb/core/persqueue/public/mlp/mlp_changer_ut.cpp

@@ -6,7 +6,7 @@ Y_UNIT_TEST_SUITE(TMLPChangerTests) {
 
 Y_UNIT_TEST(TopicNotExists) {
     auto setup = CreateSetup();
-    
+
     auto& runtime = setup->GetRuntime();
     CreateCommitterActor(runtime, {
         .DatabasePath = "/Root",
@@ -25,7 +25,7 @@ Y_UNIT_TEST(ConsumerNotExists) {
     auto setup = CreateSetup();
 
     ExecuteDDL(*setup, "CREATE TOPIC topic1");
-    
+
     auto& runtime = setup->GetRuntime();
     CreateCommitterActor(runtime, {
         .DatabasePath = "/Root",
@@ -43,7 +43,7 @@ Y_UNIT_TEST(PartitionNotExists) {
     auto setup = CreateSetup();
 
     CreateTopic(setup, "/Root/topic1", "mlp-consumer");
-    
+
     auto& runtime = setup->GetRuntime();
     CreateCommitterActor(runtime, {
         .DatabasePath = "/Root",
@@ -69,7 +69,7 @@ Y_UNIT_TEST(CommitTest) {
     setup->Write("/Root/topic1", "msg-2", 0);
 
     Sleep(TDuration::Seconds(2));
-    
+
     auto& runtime = setup->GetRuntime();
     CreateCommitterActor(runtime, {
         .DatabasePath = "/Root",
@@ -269,7 +269,7 @@ Y_UNIT_TEST(CapacityTest) {
                 hFunc(NMLP::TEvChangeResponse, Handle);
                 sFunc(TEvents::TEvPoison, PassAway);
                 sFunc(TEvents::TEvWakeup, PassAway);
-            }            
+            }
         }
 
         size_t Infly = 0;

ydb/core/persqueue/public/mlp/mlp_message_attributes.h

@@ -1,13 +1,13 @@
 #pragma once
 
+#include <ydb/core/persqueue/public/constants.h>
+
 #include <util/generic/strbuf.h>
 
 namespace NKikimr::NPQ::NMLP  {
     namespace NMessageConsts {
         constexpr TStringBuf MessageDeduplicationId = "__MessageDeduplicationId";
         constexpr TStringBuf MessageAttributes = "__MessageAttributes";
         constexpr TStringBuf DelaySeconds = "__DelaySeconds";
-        constexpr TStringBuf MessageId = "__MessageId";
-
     } // namespace NMessageConsts
 } // namespace NKikimr::NPQ::NMLP