Supported big messages by MLP reader (#27656)

Author: nshestakov

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

@@ -53,6 +53,11 @@ bool IsSucess(const TEvPQ::TEvProxyResponse::TPtr& ev) {
         ev->Get()->Response->GetErrorCode() == NPersQueue::NErrorCode::OK;
 }
 
+bool IsSucess(const TEvPersQueue::TEvResponse::TPtr& ev) {
+    return ev->Get()->Record.GetStatus() == NMsgBusProxy::MSTATUS_OK &&
+        ev->Get()->Record.GetErrorCode() == NPersQueue::NErrorCode::OK;
+}
+
 ui64 GetCookie(const TEvPQ::TEvProxyResponse::TPtr& ev) {
     return ev->Get()->Response->GetCookie();
 }

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

@@ -41,6 +41,7 @@ std::unique_ptr<TEvPQ::TEvSetClientInfo> MakeEvCommit(
 );
 
 bool IsSucess(const TEvPQ::TEvProxyResponse::TPtr& ev);
+bool IsSucess(const TEvPersQueue::TEvResponse::TPtr& ev);
 ui64 GetCookie(const TEvPQ::TEvProxyResponse::TPtr& ev);
 
 } // namespace NKikimr::NPQ::NMLP

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

@@ -221,7 +221,7 @@ void TConsumerActor::HandleOnWrite(TEvKeyValue::TEvResponse::TPtr& ev) {
 
     if (!PendingReadQueue.empty()) {
         auto msgs = std::exchange(PendingReadQueue, {});
-        RegisterWithSameMailbox(new TMessageEnricherActor(PartitionId, PartitionActorId, Config.GetName(), std::move(msgs))); // TODO excahnge
+        RegisterWithSameMailbox(new TMessageEnricherActor(TabletActorId, PartitionId, Config.GetName(), std::move(msgs)));
     }
     ReplyOk<TEvPQ::TEvMLPCommitResponse>(SelfId(), PendingCommitQueue);
     ReplyOk<TEvPQ::TEvMLPUnlockResponse>(SelfId(), PendingUnlockQueue);

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

@@ -2,10 +2,10 @@
 
 namespace NKikimr::NPQ::NMLP {
 
-TMessageEnricherActor::TMessageEnricherActor(ui32 partitionId, const TActorId& partitionActor, const TString& consumerName, std::deque<TReadResult>&& replies)
+TMessageEnricherActor::TMessageEnricherActor(const TActorId& tabletActorId, ui32 partitionId, const TString& consumerName, std::deque<TReadResult>&& replies)
     : TBaseActor(NKikimrServices::EServiceKikimr::PQ_MLP_ENRICHER)
+    , TabletActorId(tabletActorId)
     , PartitionId(partitionId)
-    , PartitionActorId(partitionActor)
     , ConsumerName(consumerName)
     , Queue(std::move(replies))
     , Backoff(5, TDuration::MilliSeconds(50))
@@ -28,22 +28,17 @@ void TMessageEnricherActor::PassAway() {
     TBase::PassAway();
 }
 
-void TMessageEnricherActor::Handle(TEvPQ::TEvProxyResponse::TPtr& ev) {
-    LOG_D("Handle TEvPQ::TEvProxyResponse");
-    if (Cookie != GetCookie(ev)) {
-        // TODO MLP
-        LOG_D("Cookie mismatch: " << Cookie << " != " << GetCookie(ev));
-        //return PassAway();
-    }
+void TMessageEnricherActor::Handle(TEvPersQueue::TEvResponse::TPtr& ev) {
+    LOG_D("Handle TEvPersQueue::TEvResponse");
 
     if (!IsSucess(ev)) {
-        LOG_W("Fetch messages failed: " << ev->Get()->Response->DebugString());
+        LOG_W("Fetch messages failed: " << ev->Get()->Record.DebugString());
         return PassAway();
     }
 
-    auto& response = ev->Get()->Response;
-    if (response->GetPartitionResponse().HasCmdReadResult()) {
-        for (auto& result : response->GetPartitionResponse().GetCmdReadResult().GetResult()) {
+    auto& response = ev->Get()->Record;
+    if (response.GetPartitionResponse().HasCmdReadResult()) {
+        for (auto& result : response.GetPartitionResponse().GetCmdReadResult().GetResult()) {
             auto offset = result.GetOffset();
 
             while(!Queue.empty()) {
@@ -54,7 +49,6 @@ void TMessageEnricherActor::Handle(TEvPQ::TEvProxyResponse::TPtr& ev) {
                 while (!reply.Offsets.empty() && offset > reply.Offsets.front()) {
                     reply.Offsets.pop_front();
                 }
-                // TODO MLP multi part messages
                 if (!reply.Offsets.empty() && offset == reply.Offsets.front()) {
                     auto* message = PendingResponse->Record.AddMessage();
                     message->MutableId()->SetPartitionId(PartitionId);
@@ -100,7 +94,7 @@ void TMessageEnricherActor::Handle(TEvents::TEvWakeup::TPtr&) {
 
 STFUNC(TMessageEnricherActor::StateWork) {
     switch (ev->GetTypeRewrite()) {
-        hFunc(TEvPQ::TEvProxyResponse, Handle);
+        hFunc(TEvPersQueue::TEvResponse, Handle);
         hFunc(TEvPQ::TEvError, Handle);
         hFunc(TEvents::TEvWakeup, Handle);
         sFunc(TEvents::TEvPoison, PassAway);
@@ -122,8 +116,17 @@ void TMessageEnricherActor::ProcessQueue() {
         auto firstOffset = reply.Offsets.front();
         auto lastOffset = Queue.back().Offsets.back();
         auto count = lastOffset - firstOffset + 1;
-        LOG_D("Fetching from offset " << firstOffset << " count " << count << " from " << PartitionActorId);
-        Send(PartitionActorId, MakeEvRead(SelfId(), ConsumerName, firstOffset, count, ++Cookie));
+        LOG_D("Fetching from offset " << firstOffset << " count " << count << " from " << TabletActorId);
+
+        auto request = std::make_unique<TEvPersQueue::TEvRequest>();
+        auto* partitionRequest = request->Record.MutablePartitionRequest();
+        partitionRequest->SetPartition(PartitionId);
+        auto* read = partitionRequest->MutableCmdRead();
+        read->SetClientId(ConsumerName);
+        read->SetOffset(firstOffset);
+        read->SetTimeoutMs(0);
+
+        Send(TabletActorId, std::move(request), 0, ++Cookie);
 
         return;
     }

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

@@ -15,16 +15,16 @@ class TMessageEnricherActor : public TBaseActor<TMessageEnricherActor>
     static constexpr TDuration Timeout = TDuration::Seconds(1);
 
 public:
-    TMessageEnricherActor(const ui32 partitionId,
-                          const TActorId& partitionActor,
+    TMessageEnricherActor(const TActorId& tabletActorId,
+                          const ui32 partitionId,
                           const TString& consumerName,
                           std::deque<TReadResult>&& replies);
 
     void Bootstrap();
     void PassAway() override;
 
 private:
-    void Handle(TEvPQ::TEvProxyResponse::TPtr&);
+    void Handle(TEvPersQueue::TEvResponse::TPtr&);
     void Handle(TEvPQ::TEvError::TPtr&);
     void Handle(TEvents::TEvWakeup::TPtr&);
 
@@ -33,8 +33,8 @@ class TMessageEnricherActor : public TBaseActor<TMessageEnricherActor>
     void ProcessQueue();
 
 private:
+    const TActorId TabletActorId;
     const ui32 PartitionId;
-    const TActorId PartitionActorId;
     const TString ConsumerName;
     std::deque<TReadResult> Queue;
     TBackoff Backoff;

ydb/core/persqueue/pqtablet/pq_impl.cpp

@@ -2598,7 +2598,6 @@ void TPersQueue::Handle(TEvPersQueue::TEvRequest::TPtr& ev, const TActorContext&
             directKey.SessionId = pipeIter->second.SessionId;
             directKey.PartitionSessionId = pipeIter->second.PartitionSessionId;
         }
-        TStringBuilder log; log << "PQ - create read proxy" << Endl;
         TActorId rr = ctx.RegisterWithSameMailbox(CreateReadProxy(ev->Sender, TabletID(), ctx.SelfID, GetGeneration(), directKey, request));
         ans = CreateResponseProxy(rr, ctx.SelfID, TopicName, p, m, s, c, ResourceMetrics, ctx);
     } else {

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

@@ -154,6 +154,34 @@ Y_UNIT_TEST_SUITE(TMLPReaderTests) {
         }
 
     }
+
+    Y_UNIT_TEST(TopicWithBigMessage) {
+        auto setup = CreateSetup();
+
+        auto bigMessage = NUnitTest::RandomString(1_MB);
+
+        CreateTopic(setup, "/Root/topic1", "mlp-consumer");
+        setup->Write("/Root/topic1", bigMessage, 0);
+
+        auto& runtime = setup->GetRuntime();
+        CreateReaderActor(runtime, {
+            .DatabasePath = "/Root",
+            .TopicName = "/Root/topic1",
+            .Consumer = "mlp-consumer",
+            .WaitTime = TDuration::Seconds(3),
+            .VisibilityTimeout = TDuration::Seconds(30),
+            .MaxNumberOfMessage = 1,
+            .UncompressMessages = true
+        });
+
+        auto response = GetReadResponse(runtime);
+        UNIT_ASSERT_VALUES_EQUAL(response->Messages.size(), 1);
+        UNIT_ASSERT_VALUES_EQUAL(response->Messages[0].MessageId.PartitionId, 0);
+        UNIT_ASSERT_VALUES_EQUAL(response->Messages[0].MessageId.Offset, 0);
+        UNIT_ASSERT_VALUES_EQUAL(response->Messages[0].Data, bigMessage);
+    }
+
+
 }
 
 } // namespace NKikimr::NPQ::NMLP

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

@@ -29,7 +29,7 @@ inline auto CreateSetup() {
             NKikimrServices::PERSQUEUE,
             NKikimrServices::PERSQUEUE_READ_BALANCER,
         },
-        NActors::NLog::PRI_INFO
+        NActors::NLog::PRI_DEBUG
     );
     return setup;
 }