[YDB_LOG] Migrate ydb/core/persqueue/prqb (#45807)

10 files changed, 476 insertions, 192 deletions

diff --git a/ydb/core/persqueue/pqrb/mirror_describer.cpp b/ydb/core/persqueue/pqrb/mirror_describer.cpp
index 47185c8846d..6cd1e3e3687 100644
--- a/ydb/core/persqueue/pqrb/mirror_describer.cpp
+++ b/ydb/core/persqueue/pqrb/mirror_describer.cpp
@@ -4,6 +4,8 @@
 
 #include <google/protobuf/util/message_differencer.h>
 
+#define YDB_LOG_THIS_FILE_COMPONENT Service
+
 #define PQ_ENSURE(condition) AFL_ENSURE(condition)("topic", TopicName)
 
 using namespace NPersQueue;
@@ -34,7 +36,8 @@ void TMirrorDescriber::StartInit(const TActorContext& ctx) {
 }
 
 void TMirrorDescriber::Handle(TEvents::TEvPoisonPill::TPtr&, const TActorContext& ctx) {
-    LOG_N("killed");
+    YDB_LOG_NOTICE("Killed",
+        {"logPrefix", NPQ_LOG_PREFIX});
     CredentialsProvider = nullptr;
     Die(ctx);
 }
@@ -44,10 +47,13 @@ void TMirrorDescriber::HandleChangeConfig(TEvPQ::TEvChangePartitionConfig::TPtr&
         Config,
         ev->Get()->Config.GetPartitionConfig().GetMirrorFrom()
     );
-    LOG_D("got new config, equal with previous: " << equalConfigs);
+    YDB_LOG_DEBUG("Got new config, equal with",
+        {"logPrefix", NPQ_LOG_PREFIX},
+        {"previous", equalConfigs});
     if (!equalConfigs) {
         Config = ev->Get()->Config.GetPartitionConfig().GetMirrorFrom();
-        LOG_I("changing config");
+        YDB_LOG_INFO("Changing config",
+            {"logPrefix", NPQ_LOG_PREFIX});
         StartInit(ctx);
     }
 }
@@ -56,13 +62,17 @@ void TMirrorDescriber::HandleDescriptionResult(TEvPQ::TEvMirrorTopicDescription:
     DescribeTopicRequestInFlight = false;
     const auto& description = ev->Get()->Description;
     if (!description.has_value()) {
-        LOG_E("cannot describe topic " << description.error());
+        YDB_LOG_ERROR("Cannot describe topic",
+            {"logPrefix", NPQ_LOG_PREFIX},
+            {"error", description.error()});
         ScheduleWithIncreasingTimeout<TEvents::TEvWakeup>(SelfId(), DescribeRetryTimeout, DESCRIBE_RETRY_TIMEOUT_MAX, ctx);
         return;
     }
     const NYdb::NTopic::TDescribeTopicResult& result = description.value();
     if (!result.IsSuccess()) {
-        LOG_E("cannot describe topic " << result.GetIssues().ToString());
+        YDB_LOG_ERROR("Cannot describe topic",
+            {"logPrefix", NPQ_LOG_PREFIX},
+            {"issues", result.GetIssues()});
         ScheduleWithIncreasingTimeout<TEvents::TEvWakeup>(SelfId(), DescribeRetryTimeout, DESCRIBE_RETRY_TIMEOUT_MAX, ctx);
         return;
     }
@@ -71,14 +81,17 @@ void TMirrorDescriber::HandleDescriptionResult(TEvPQ::TEvMirrorTopicDescription:
         descr.SerializeTo(req);
         return req.ShortUtf8DebugString();
     };
-    LOG_T("topic description: " << debugTopicDescriptionString(description.value().GetTopicDescription()));
+    YDB_LOG_TRACE("Topic",
+        {"logPrefix", NPQ_LOG_PREFIX},
+        {"description", debugTopicDescriptionString(description.value().GetTopicDescription())});
     ctx.Send(TabletActorId, ev->Release());
     ctx.Schedule(DESCRIBE_RETRY_TIMEOUT_MAX, new TEvents::TEvWakeup());
 }
 
 void TMirrorDescriber::DescribeTopic(const TActorContext& ctx) {
     if (DescribeTopicRequestInFlight) {
-        LOG_I("description request already inflight.");
+        YDB_LOG_INFO("Description request already inflight",
+            {"logPrefix", NPQ_LOG_PREFIX});
         return;
     }
 
@@ -111,7 +124,8 @@ void TMirrorDescriber::DescribeTopic(const TActorContext& ctx) {
 
 void TMirrorDescriber::HandleInitCredentials(TEvPQ::TEvInitCredentials::TPtr& /*ev*/, const TActorContext& ctx) {
     if (CredentialsRequestInFlight) {
-        LOG_W("credentials request already inflight.");
+        YDB_LOG_WARN("Credentials request already inflight",
+            {"logPrefix", NPQ_LOG_PREFIX});
         return;
     }
     CredentialsProvider = nullptr;
@@ -145,13 +159,17 @@ void TMirrorDescriber::HandleInitCredentials(TEvPQ::TEvInitCredentials::TPtr& /*
 void TMirrorDescriber::HandleCredentialsCreated(TEvPQ::TEvCredentialsCreated::TPtr& ev, const TActorContext& ctx) {
     CredentialsRequestInFlight = false;
     if (ev->Get()->Error) {
-        LOG_W("cannot initialize credentials provider: " << ev->Get()->Error.value());
+        YDB_LOG_WARN("Cannot initialize credentials",
+            {"logPrefix", NPQ_LOG_PREFIX},
+            {"provider", ev->Get()->Error.value()});
         ScheduleWithIncreasingTimeout<TEvPQ::TEvInitCredentials>(SelfId(), CredentialsInitInterval, INIT_INTERVAL_MAX, ctx);
         return;
     }
 
     CredentialsProvider = ev->Get()->Credentials;
-    LOG_N("credentials provider created " << bool(CredentialsProvider));
+    YDB_LOG_NOTICE("Credentials provider created",
+        {"logPrefix", NPQ_LOG_PREFIX},
+        {"hasCredentialsProvider", bool(CredentialsProvider)});
     CredentialsInitInterval = INIT_INTERVAL_START;
     ScheduleDescription(ctx);
 }
diff --git a/ydb/core/persqueue/pqrb/partition_scale_manager.cpp b/ydb/core/persqueue/pqrb/partition_scale_manager.cpp
index 12b49163dbc..517fe6591ba 100644
--- a/ydb/core/persqueue/pqrb/partition_scale_manager.cpp
+++ b/ydb/core/persqueue/pqrb/partition_scale_manager.cpp
@@ -6,6 +6,8 @@
 #include <algorithm>
 #include <ranges>
 
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
+
 namespace NKikimr {
 namespace NPQ {
 
@@ -35,9 +37,13 @@ void TPartitionScaleManager::HandleScaleStatusChange(const ui32 partitionId, NKi
     TMaybe<NKikimrPQ::TPartitionScaleParticipants> participants,
     TMaybe<TString> splitBoundary,
     const TActorContext& ctx) {
-    PQ_LOG_D("Handle HandleScaleStatusChange. Scale status: " << NKikimrPQ::EScaleStatus_Name(scaleStatus));
+    YDB_LOG_DEBUG("Handle HandleScaleStatusChange. Scale",
+        {"logPrefix", LogPrefix()},
+        {"status", NKikimrPQ::EScaleStatus_Name(scaleStatus)});
     if (scaleStatus == NKikimrPQ::EScaleStatus::NEED_SPLIT) {
-        PQ_LOG_D("::HandleScaleStatusChange need to split partition " << partitionId);
+        YDB_LOG_DEBUG("::HandleScaleStatusChange need to split partition",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId});
         TPartitionScaleOperationInfo op{
             .PartitionId = partitionId,
             .PartitionScaleParticipants = std::move(participants),
@@ -58,13 +64,15 @@ void TPartitionScaleManager::TrySendScaleRequest(const TActorContext& ctx) {
 
     auto splitMergeRequest = BuildScaleRequest(ctx);
     if (splitMergeRequest.Empty()) {
-        PQ_LOG_D("splitMergeRequest empty");
+        YDB_LOG_DEBUG("SplitMergeRequest empty",
+            {"logPrefix", LogPrefix()});
         return;
     }
 
     RequestInflight = true;
     RootPartitionsResetRequestInflight = !splitMergeRequest.SetBoundary.empty();
-    PQ_LOG_D("send split request");
+    YDB_LOG_DEBUG("Send split request",
+        {"logPrefix", LogPrefix()});
     CurrentScaleRequest = ctx.Register(new TPartitionScaleRequest(
         TopicName,
         TopicPath,
@@ -129,12 +137,12 @@ TPartitionScaleManager::TScaleRequest TPartitionScaleManager::BuildScaleRequest(
     auto mergesToApply = BuildMergeRequest(allowedSplitsCount);
     auto splitsToApply = BuildSplitRequest(allowedSplitsCount);
 
-    PQ_LOG_D(fmt::format("Scale request: #splits={}, #unprocessed={}, splitsLimit={}, #merges={}",
-        splitsToApply.Requests.size(),
-        splitsToApply.Unprocessed,
-        allowedSplitsCountLimit,
-        mergesToApply.Requests.size()
-    ));
+    YDB_LOG_DEBUG("Scale request",
+        {"logPrefix", LogPrefix()},
+        {"splits", splitsToApply.Requests.size()},
+        {"unprocessed", splitsToApply.Unprocessed},
+        {"splitsLimit", allowedSplitsCountLimit},
+        {"merges", mergesToApply.Requests.size()});
 
     return {
         .Split = std::move(splitsToApply.Requests),
@@ -166,19 +174,20 @@ TPartitionScaleManager::TRequests<TPartitionScaleManager::TPartitionBoundary> TP
                 allowedSplitsCount -= cost;
                 boundsToApply.push_back(std::move(part));
             } else {
-                PQ_LOG_W(fmt::format("MaxActivePartitions ({}) is too low to recreate {} root partitions from the mirror source topic",
-                                     BalancerConfig.MaxActivePartitions,
-                                     RootPartitionsToCreate->size()));
+                YDB_LOG_WARN("MaxActivePartitions is too low to recreate root partitions from the mirror source topic",
+                    {"logPrefix", LogPrefix()},
+                    {"maxActivePartitions", BalancerConfig.MaxActivePartitions},
+                    {"createPartitions", RootPartitionsToCreate->size()});
                 // don't send request at all, if there is not enough quota
                 return {
                     .Unprocessed = RootPartitionsToCreate->size(),
                 };
             }
         }
-        PQ_LOG_D(fmt::format("Set partition boundaries requsts: #modify={}, #create{}",
-            modifyPartitions,
-            createPartitions
-        ));
+        YDB_LOG_DEBUG("Set partition boundaries requsts",
+            {"logPrefix", LogPrefix()},
+            {"modify", modifyPartitions},
+            {"create", createPartitions});
     }
     return {
         .Requests = std::move(boundsToApply),
@@ -221,21 +230,29 @@ TPartitionScaleManager::TBuildSplitScaleRequestResult TPartitionScaleManager::Bu
     const ui32 partitionId = splitParameters.PartitionId;
     if (MirroredFromSomewhere)  {
         if (!AppData()->FeatureFlags.GetEnableMirroredTopicSplitMerge()) {
-            PQ_LOG_D("split request for mirrored topic is disabled. Partition# " << partitionId);
+            YDB_LOG_DEBUG("Split request for mirrored topic is disabled",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = false};
         }
         if (!splitParameters.PartitionScaleParticipants.Defined()) {
-            PQ_LOG_NOTICE("split request for mirrored topic doesn't have prescribed partition ids. Partition# " << partitionId);
+            YDB_LOG_NOTICE("Split request for mirrored topic doesn't have prescribed partition ids",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = true};
         }
     }
     const auto* node = PartitionGraph.GetPartition(partitionId);
     if (node == nullptr) {
         if (splitParameters.PartitionScaleParticipants.Defined()) {
-            PQ_LOG_NOTICE("attempt to split partition that was not created yet. Partition# " << partitionId);
+            YDB_LOG_NOTICE("Attempt to split partition that was not created yet",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = false};
         } else {
-            PQ_LOG_ERROR("partition not found. Partition# " << partitionId);
+            YDB_LOG_ERROR("Partition not found",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = true};
         }
     }
@@ -244,18 +261,25 @@ TPartitionScaleManager::TBuildSplitScaleRequestResult TPartitionScaleManager::Bu
         auto to = node->To;
         auto mid = splitParameters.SplitBoundary.GetOrElse(MiddleOf(from, to));
         if (mid.empty()) {
-            PQ_LOG_ERROR("wrong partition key range. Can't get mid. Partition# " << partitionId);
+            YDB_LOG_ERROR("Wrong partition key range. Can't get mid",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = true};
         }
 
         if (splitParameters.PartitionScaleParticipants.Defined() && splitParameters.PartitionScaleParticipants->AdjacentPartitionIdsSize() != 0) {
-            PQ_LOG_ERROR("split request cannot have adjacent partitions. Partition# " << partitionId);
+            YDB_LOG_ERROR("Split request cannot have adjacent partitions",
+                {"logPrefix", LogPrefix()},
+                {"partition", partitionId});
             return {.Split = Nothing(), .Remove = true};
         }
 
-        PQ_LOG_D("partition split ranges. From# '" << ToHex(from)
-                << "'. To# '" << ToHex(to) << "'. Mid# '" << ToHex(mid)
-                << "'. Partition# " << partitionId);
+        YDB_LOG_DEBUG("Partition split ranges",
+            {"logPrefix", LogPrefix()},
+            {"fromHex", ToHex(from)},
+            {"toHex", ToHex(to)},
+            {"midHex", ToHex(mid)},
+            {"partition", partitionId});
 
         TPartitionSplit split;
         split.set_partition(partitionId);
@@ -264,7 +288,10 @@ TPartitionScaleManager::TBuildSplitScaleRequestResult TPartitionScaleManager::Bu
             for (const auto& childPartitionId : splitParameters.PartitionScaleParticipants->GetChildPartitionIds()) {
                 split.add_childpartitionids(childPartitionId);
                 if (const auto* childNode = PartitionGraph.GetPartition(childPartitionId); childNode != nullptr) {
-                    PQ_LOG_NOTICE(fmt::format("Child partition# {} already exists. Performing unordered split. Partition# {}", childPartitionId, partitionId));
+                    YDB_LOG_NOTICE("Child partition already exists. Performing unordered split",
+                        {"logPrefix", LogPrefix()},
+                        {"childPartition", childPartitionId},
+                        {"partition", partitionId});
                     split.set_createrootlevelsibling(true);
                 }
             }
@@ -276,7 +303,10 @@ TPartitionScaleManager::TBuildSplitScaleRequestResult TPartitionScaleManager::Bu
             const auto& prescribedChildrenIds = splitParameters.PartitionScaleParticipants->GetChildPartitionIds();
             if (!std::ranges::is_permutation(nodeChildrenIds, prescribedChildrenIds)) {
                 const std::string mappingStr = fmt::format("([{}]->[{}])", fmt::join(nodeChildrenIds, ","), fmt::join(prescribedChildrenIds, ","));
-                PQ_LOG_ERROR("trying to split partition into different set of children partitions " << mappingStr << ". Partition# " << partitionId);
+                YDB_LOG_ERROR("Trying to split partition into different set of children partitions",
+                    {"logPrefix", LogPrefix()},
+                    {"mappingStr", mappingStr},
+                    {"partition", partitionId});
             }
         }
         return {.Split = Nothing(), .Remove = true};
@@ -287,7 +317,9 @@ void TPartitionScaleManager::HandleScaleRequestResult(TPartitionScaleRequest::TE
     RequestInflight = false;
     LastResponseTime = ctx.Now();
     auto result = ev->Get();
-    PQ_LOG_D("HandleScaleRequestResult scale request result: " << result->Status);
+    YDB_LOG_DEBUG("HandleScaleRequestResult scale request",
+        {"logPrefix", LogPrefix()},
+        {"result", result->Status});
     if (result->Status == TEvTxUserProxy::TResultStatus::ExecComplete) {
         RequestTimeout = TDuration::Zero();
         Backoff.Reset();
@@ -312,24 +344,27 @@ void TPartitionScaleManager::UpdateMirrorRootPartitionsSet() {
 
     NMirror::TMirrorGraphComparisonResult cmp = NMirror::ComparePartitionGraphs(PartitionGraph, MirrorTopicDescription->GetPartitions());
     if (!cmp.RootPartitionsMismatch.has_value()) {
-        PQ_LOG_D("Topic has all root partitions from the source topic");
+        YDB_LOG_DEBUG("Topic has all root partitions from the source topic",
+            {"logPrefix", LogPrefix()});
         RootPartitionsToCreate.reset();
         MirrorTopicError.reset();
         return;
     }
     auto& rootPartitionsMismatch = cmp.RootPartitionsMismatch.value();
     if (rootPartitionsMismatch.Error.has_value()) {
-        std::string msg = TStringBuilder() << "Incompatable configuration of root partitions between source and target topics:" << rootPartitionsMismatch.Error.value();
-        PQ_LOG_ERROR(msg);
+        YDB_LOG_ERROR("Incompatable configuration of root partitions between source and target topics",
+            {"logPrefix", LogPrefix()},
+            {"topics", rootPartitionsMismatch.Error.value()});
         RootPartitionsToCreate.reset();
         MirrorTopicError = std::move(*rootPartitionsMismatch.Error);
         return;
     }
     const size_t existingPartitions = std::ranges::count(rootPartitionsMismatch.AlterRootPartitions, NMirror::EPartitionAction::Modify, &NMirror::TPartitionWithBounds::Action);
     const size_t newPartitions = std::ranges::count(rootPartitionsMismatch.AlterRootPartitions, NMirror::EPartitionAction::Create, &NMirror::TPartitionWithBounds::Action);
-    PQ_LOG_I(fmt::format("Topic has less root partitions than the mirror source. New configuration has {}+{} partitions.",
-                         existingPartitions,
-                         newPartitions));
+    YDB_LOG_INFO("Topic has less root partitions than the mirror source",
+        {"logPrefix", LogPrefix()},
+        {"existing", existingPartitions},
+        {"new", newPartitions});
 
     RootPartitionsToCreate = std::move(rootPartitionsMismatch.AlterRootPartitions);
     MirrorTopicError.reset();
@@ -341,7 +376,8 @@ std::expected<void, std::string> TPartitionScaleManager::HandleMirrorTopicDescri
     } else {
         auto& description = ev->Get()->Description;
         if (!description.has_value() || !description.value().IsSuccess()) {
-            PQ_LOG_W("Ignoring invalid mirror source description");
+            YDB_LOG_WARN("Ignoring invalid mirror source description",
+                {"logPrefix", LogPrefix()});
             return {};
         }
         MirrorTopicDescription.emplace(std::move(description->GetTopicDescription()));
diff --git a/ydb/core/persqueue/pqrb/partition_scale_request.cpp b/ydb/core/persqueue/pqrb/partition_scale_request.cpp
index e09adfe7b0e..30d1298c230 100644
--- a/ydb/core/persqueue/pqrb/partition_scale_request.cpp
+++ b/ydb/core/persqueue/pqrb/partition_scale_request.cpp
@@ -3,6 +3,8 @@
 
 #include <ydb/core/protos/schemeshard/operations.pb.h>
 
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
+
 namespace NKikimr {
 namespace NPQ {
 
@@ -76,7 +78,8 @@ void TPartitionScaleRequest::FillProposeRequest(TEvTxUserProxy::TEvProposeTransa
         }
         logMessage << ".";
     }
-    PQ_LOG_D( logMessage);
+    YDB_LOG_DEBUG(logMessage,
+        {"logPrefix", LogPrefix()});
 
     for(const auto& merge: Merges) {
         auto* newMerge = groupDescription.AddMerge();
@@ -124,7 +127,9 @@ void TPartitionScaleRequest::Handle(TEvTxUserProxy::TEvProposeTransactionStatus:
         for (auto& issue : ev->Get()->Record.GetIssues()) {
             issues << issue.ShortDebugString() + ", ";
         }
-        PQ_LOG_ERROR("TPartitionScaleRequest SchemaShard error when trying to execute a split request: " << issues);
+        YDB_LOG_ERROR("TPartitionScaleRequest SchemaShard error when trying to execute a split",
+            {"logPrefix", LogPrefix()},
+            {"request", issues});
         Send(ParentActorId, scaleRequestResult.release());
         Die(ctx);
     } else {
diff --git a/ydb/core/persqueue/pqrb/read_balancer.cpp b/ydb/core/persqueue/pqrb/read_balancer.cpp
index 08ad274f8de..5eafeb10d21 100644
--- a/ydb/core/persqueue/pqrb/read_balancer.cpp
+++ b/ydb/core/persqueue/pqrb/read_balancer.cpp
@@ -16,6 +16,8 @@
 #include <library/cpp/string_utils/base64/base64.h>
 #include <library/cpp/random_provider/random_provider.h>
 
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
+
 #define PQ_ENSURE(condition) AFL_ENSURE(condition)("tablet_id", TabletID())("path", Path)("topic", Topic)
 
 namespace NKikimr {
@@ -142,7 +144,9 @@ void TPersQueueReadBalancer::InitDone(const TActorContext &ctx) {
         }
         return s;
     };
-    PQ_LOG_D(getInitLog());
+    YDB_LOG_DEBUG("BALANCER INIT DONE dump logPrefix, getInitLog",
+        {"logPrefix", LogPrefix()},
+        {"getInitLog", getInitLog()});
 
     for (auto &ev : UpdateEvents) {
         ctx.Send(ctx.SelfID, ev.Release());
@@ -161,7 +165,8 @@ void TPersQueueReadBalancer::InitDone(const TActorContext &ctx) {
 }
 
 void TPersQueueReadBalancer::HandleWakeup(TEvents::TEvWakeup::TPtr& ev, const TActorContext &ctx) {
-    PQ_LOG_D("TPersQueueReadBalancer::HandleWakeup");
+    YDB_LOG_DEBUG("TPersQueueReadBalancer::HandleWakeup",
+        {"logPrefix", LogPrefix()});
 
     switch (ev->Get()->Tag) {
         case TPartitionScaleManager::TRY_SCALE_REQUEST_WAKE_UP_TAG: {
@@ -209,9 +214,13 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvUpdateBalancerConfig::TPtr
         if (!WaitingResponse.empty()) { //got transaction infly
             WaitingResponse.push_back(ev->Sender);
         } else { //version already applied
-            PQ_LOG_D("BALANCER Topic " << Topic << "Tablet " << TabletID()
-                    << " Config already applied version " << record.GetVersion() << " actor " << ev->Sender
-                    << " txId " << record.GetTxId());
+            YDB_LOG_DEBUG("BALANCER Topic Tablet Config already applied version actor txId",
+                {"logPrefix", LogPrefix()},
+                {"topic", Topic},
+                {"tabletID", TabletID()},
+                {"version", record.GetVersion()},
+                {"sender", ev->Sender},
+                {"txId", record.GetTxId()});
             THolder<TEvPersQueue::TEvUpdateConfigResponse> res{new TEvPersQueue::TEvUpdateConfigResponse};
             res->Record.SetStatus(NKikimrPQ::OK);
             res->Record.SetTxId(record.GetTxId());
@@ -374,7 +383,9 @@ TStringBuilder TPersQueueReadBalancer::LogPrefix() const {
 void TPersQueueReadBalancer::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActorContext& ctx)
 {
     auto tabletId = ev->Get()->TabletId;
-    PQ_LOG_D("TEvClientDestroyed " << tabletId);
+    YDB_LOG_DEBUG("TEvClientDestroyed",
+        {"logPrefix", LogPrefix()},
+        {"tabletId", tabletId});
 
     ClosePipe(tabletId, ctx);
     RequestTabletIfNeeded(tabletId, ctx, true);
@@ -391,7 +402,10 @@ void TPersQueueReadBalancer::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev,
         ClosePipe(ev->Get()->TabletId, ctx);
         RequestTabletIfNeeded(ev->Get()->TabletId, ctx, true);
 
-        PQ_LOG_ERROR("TEvClientConnected Status " << ev->Get()->Status << ", TabletId " << tabletId);
+        YDB_LOG_ERROR("TEvClientConnected Status TabletId",
+            {"logPrefix", LogPrefix()},
+            {"status", ev->Get()->Status},
+            {"tabletId", tabletId});
         return;
     }
 
@@ -402,10 +416,16 @@ void TPersQueueReadBalancer::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev,
         it->second.Generation = ev->Get()->Generation;
         it->second.NodeId = ev->Get()->ServerId.NodeId();
 
-        PQ_LOG_D("TEvClientConnected TabletId " << tabletId << ", NodeId " << ev->Get()->ServerId.NodeId() << ", Generation " << ev->Get()->Generation);
+        YDB_LOG_DEBUG("TEvClientConnected TabletId NodeId Generation",
+            {"logPrefix", LogPrefix()},
+            {"tabletId", tabletId},
+            {"nodeId", ev->Get()->ServerId.NodeId()},
+            {"generation", ev->Get()->Generation});
     }
     else
-        PQ_LOG_I("TEvClientConnected Pipe is not found, TabletId " << tabletId);
+        YDB_LOG_INFO("TEvClientConnected Pipe is not found, TabletId",
+            {"logPrefix", LogPrefix()},
+            {"tabletId", tabletId});
 }
 
 void TPersQueueReadBalancer::ClosePipe(const ui64 tabletId, const TActorContext& ctx)
@@ -453,7 +473,10 @@ void TPersQueueReadBalancer::RequestTabletIfNeeded(const ui64 tabletId, const TA
             StatsRequestTracker.Cookies[tabletId] = cookie;
         }
 
-        PQ_LOG_D("Send TEvPersQueue::TEvStatus TabletId: " << tabletId << " Cookie: " << cookie);
+        YDB_LOG_DEBUG("Send TEvPersQueue::TEvStatus",
+            {"logPrefix", LogPrefix()},
+            {"tabletId", tabletId},
+            {"cookie", cookie});
         NTabletPipe::SendData(ctx, pipeClient, new TEvPersQueue::TEvStatus("", true), cookie);
     }
 }
@@ -531,11 +554,13 @@ void TPersQueueReadBalancer::CheckStat(const TActorContext& ctx) {
     UpdateCounters(ctx);
 
     TEvPersQueue::TEvPeriodicTopicStats* ev = GetStatsEvent();
-    PQ_LOG_D("Send TEvPeriodicTopicStats PathId: " << PathId
-            << " Generation: " << Generation
-            << " StatsReportRound: " << StatsReportRound
-            << " DataSize: " << TopicMetricsHandler->GetTopicMetrics().TotalDataSize
-            << " UsedReserveSize: " << TopicMetricsHandler->GetTopicMetrics().TotalUsedReserveSize);
+    YDB_LOG_DEBUG("Send TEvPeriodicTopicStats",
+        {"logPrefix", LogPrefix()},
+        {"pathId", PathId},
+        {"generation", Generation},
+        {"statsReportRound", StatsReportRound},
+        {"dataSize", TopicMetricsHandler->GetTopicMetrics().TotalDataSize},
+        {"usedReserveSize", TopicMetricsHandler->GetTopicMetrics().TotalUsedReserveSize});
 
     NTabletPipe::SendData(ctx, GetPipeClient(SchemeShardId, ctx), ev);
 
@@ -622,8 +647,12 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvGetPartitionsLocation::TPtr
         pResponse->SetNodeId(iter->second.NodeId.GetRef());
         pResponse->SetGeneration(iter->second.Generation.GetRef());
 
-        PQ_LOG_D("The partition location was added to response: TabletId " << tabletId << ", PartitionId " << partitionId
-                << ", NodeId " << pResponse->GetNodeId() << ", Generation " << pResponse->GetGeneration());
+        YDB_LOG_DEBUG("The partition location was added to response: TabletId PartitionId NodeId Generation",
+            {"logPrefix", LogPrefix()},
+            {"tabletId", tabletId},
+            {"partitionId", partitionId},
+            {"pResponseNodeId", pResponse->GetNodeId()},
+            {"pResponseGeneration", pResponse->GetGeneration()});
 
         return true;
     };
@@ -713,7 +742,10 @@ void TPersQueueReadBalancer::Handle(NSchemeShard::TEvSchemeShard::TEvSubDomainPa
     if (SchemeShardId == msg->SchemeShardId &&
        (!SubDomainPathId || SubDomainPathId->OwnerId != msg->SchemeShardId))
     {
-        PQ_LOG_D("Discovered subdomain " << msg->LocalPathId << " at RB " << TabletID());
+        YDB_LOG_DEBUG("Discovered subdomain at RB",
+            {"logPrefix", LogPrefix()},
+            {"localPathId", msg->LocalPathId},
+            {"tabletID", TabletID()});
 
         SubDomainPathId.emplace(msg->SchemeShardId, msg->LocalPathId);
         Execute(new TTxWriteSubDomainPathId(this), ctx);
@@ -767,8 +799,11 @@ void TPersQueueReadBalancer::Handle(TEvTxProxySchemeCache::TEvWatchNotifyUpdated
             .GetDomainState()
             .GetDiskQuotaExceeded();
 
-        PQ_LOG_D("Discovered subdomain " << msg->PathId << " state, outOfSpace = " << outOfSpace
-                << " at RB " << TabletID());
+        YDB_LOG_DEBUG("Discovered subdomain state, at RB",
+            {"logPrefix", LogPrefix()},
+            {"pathId", msg->PathId},
+            {"outOfSpace", outOfSpace},
+            {"tabletID", TabletID()});
 
         SubDomainOutOfSpace = outOfSpace;
 
@@ -847,7 +882,9 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvGetReadSessionsInfo::TPtr&
 void TPersQueueReadBalancer::Handle(TEvPQ::TEvMLPConsumerStatus::TPtr& ev, const TActorContext& ctx)
 {
     Y_UNUSED(ctx);
-    PQ_LOG_D("Handle TEvPQ::TEvMLPConsumerStatus " << ev->Get()->Record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPQ::TEvMLPConsumerStatus",
+        {"logPrefix", LogPrefix()},
+        {"ev", ev->Get()->Record.ShortDebugString()});
     MLPBalancer->Handle(ev);
 }
 
@@ -874,13 +911,16 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvBalancingUnsubscribe::TPtr&
 
 void TPersQueueReadBalancer::Handle(TEvPQ::TEvPartitionScaleStatusChanged::TPtr& ev, const TActorContext& ctx) {
     if (!SplitMergeEnabled(TabletConfig)) {
-        PQ_LOG_D("Skip TEvPartitionScaleStatusChanged: autopartitioning disabled.");
+        YDB_LOG_DEBUG("Skip TEvPartitionScaleStatusChanged: autopartitioning disabled",
+            {"logPrefix", LogPrefix()});
         return;
     }
     auto& record = ev->Get()->Record;
     auto* node = PartitionGraph.GetPartition(record.GetPartitionId());
     if (!node) {
-        PQ_LOG_D("Skip TEvPartitionScaleStatusChanged: partition " << record.GetPartitionId() << " not found.");
+        YDB_LOG_DEBUG("Skip TEvPartitionScaleStatusChanged: partition not found",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", record.GetPartitionId()});
         return;
     }
 
@@ -893,7 +933,8 @@ void TPersQueueReadBalancer::Handle(TEvPQ::TEvPartitionScaleStatusChanged::TPtr&
             ctx
         );
     } else {
-        PQ_LOG_NOTICE("Skip TEvPartitionScaleStatusChanged: scale manager isn`t initialized.");
+        YDB_LOG_NOTICE("Skip TEvPartitionScaleStatusChanged: scale manager isn`t initialized",
+            {"logPrefix", LogPrefix()});
     }
 }
 
@@ -907,7 +948,8 @@ void TPersQueueReadBalancer::Handle(TPartitionScaleRequest::TEvPartitionScaleReq
 }
 
 void TPersQueueReadBalancer::Handle(TEvPQ::TEvMirrorTopicDescription::TPtr& ev, const TActorContext& ctx) {
-    PQ_LOG_D("Received TEvMirrorTopicDescription");
+    YDB_LOG_DEBUG("Received TEvMirrorTopicDescription",
+        {"logPrefix", LogPrefix()});
     if (!MirroringEnabled(TabletConfig)) {
         return;
     }
@@ -928,13 +970,17 @@ void TPersQueueReadBalancer::BroadcastPartitionError(const TString& message, con
 }
 
 void TPersQueueReadBalancer::Handle(TEvPQ::TEvMLPGetPartitionRequest::TPtr& ev) {
-    PQ_LOG_D("Handle TEvPQ::TEvMLPGetPartitionRequest: " << ev->Get()->Record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPQ::TEvMLPGetPartitionRequest",
+        {"logPrefix", LogPrefix()},
+        {"ev", ev->Get()->Record.ShortDebugString()});
     PendingMLPGetPartitionRequests.push_back(std::move(ev));
     ProcessMLPGetPartitionRequests(ActorContext());
 }
 
 void TPersQueueReadBalancer::Handle(TEvPQ::TEvMLPGetRuntimeAttributesRequest::TPtr& ev) {
-    PQ_LOG_D("Handle TEvPQ::TEvMLPGetRuntimeAttributesRequest: " << ev->Get()->Record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle",
+        {"logPrefix", LogPrefix()},
+        {"mlpGetRuntimeAttributesRequest", ev->Get()->Record.ShortDebugString()});
     if (StatsRequestTracker.StatsReceived) {
         return MLPBalancer->Handle(ev);
     }
diff --git a/ydb/core/persqueue/pqrb/read_balancer__balancing.cpp b/ydb/core/persqueue/pqrb/read_balancer__balancing.cpp
index 98ba0856039..b91d5afe224 100644
--- a/ydb/core/persqueue/pqrb/read_balancer__balancing.cpp
+++ b/ydb/core/persqueue/pqrb/read_balancer__balancing.cpp
@@ -3,8 +3,7 @@
 
 #include <ydb/core/persqueue/public/utils.h>
 
-#define DEBUG(message)
-
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
 
 namespace NKikimr::NPQ::NBalancing {
 
@@ -164,17 +163,23 @@ TString TPartitionFamily::LogPrefix() const {
 
 void TPartitionFamily::Release(const TActorContext& ctx, ETargetStatus targetStatus) {
     if (Status != EStatus::Active) {
-        PQ_LOG_CRIT("releasing the family " << DebugStr() << " that isn't active");
+        YDB_LOG_CRIT("Releasing the family that isn't active",
+            {"logPrefix", LogPrefix()},
+            {"debugStr", DebugStr()});
         return;
     }
 
     if (!Session) {
-        PQ_LOG_CRIT("releasing the family " << DebugStr() << " that does not have a session");
+        YDB_LOG_CRIT("Releasing the family that does not have a session",
+            {"logPrefix", LogPrefix()},
+            {"debugStr", DebugStr()});
         return;
     }
 
-    PQ_LOG_I(" release partitions [" << JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end())
-            << "]. Target status " << targetStatus);
+    YDB_LOG_INFO("Release partitions. Target status",
+        {"logPrefix", LogPrefix()},
+        {"lockedPartitions", JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end())},
+        {"targetStatus", targetStatus});
 
     Status = EStatus::Releasing;
     TargetStatus = targetStatus;
@@ -193,25 +198,35 @@ void TPartitionFamily::Release(const TActorContext& ctx, ETargetStatus targetSta
 
 bool TPartitionFamily::Unlock(const TActorId& sender, ui32 partitionId, const TActorContext& ctx) {
     if (!Session || Session->Pipe != sender) {
-        PQ_LOG_D("try unlock the partition " << partitionId << " from other sender");
+        YDB_LOG_DEBUG("Try unlock the partition from other sender",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId});
         return false;
     }
 
     if (Status != EStatus::Releasing) {
-        PQ_LOG_CRIT("try unlock partition " << partitionId << " but family status is " << Status);
+        YDB_LOG_CRIT("Try unlock partition but family status is",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"status", Status});
         return false;
     }
 
     if (!LockedPartitions.erase(partitionId)) {
-        PQ_LOG_CRIT("try unlock partition " << partitionId << " but partition isn't locked."
-                << " Locked partitions are [" << JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end()) << "]");
+        YDB_LOG_CRIT("Try unlock partition but partition isn't locked. Locked partitions are",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"lockedPartitions", JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end())});
         return false;
     }
 
     --Session->ReleasingPartitionCount;
 
     if (!LockedPartitions.empty()) {
-        PQ_LOG_D("partition " << partitionId << " was unlocked but wait else [" << JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end()) << "]");
+        YDB_LOG_DEBUG("Partition was unlocked, but wait",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"lockedPartitions", JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end())});
         return false;
     }
 
@@ -238,7 +253,8 @@ bool TPartitionFamily::Reset(ETargetStatus targetStatus, const TActorContext& ct
             return false;
 
         case ETargetStatus::Free:
-            PQ_LOG_T(" is free.");
+            YDB_LOG_TRACE("Is free",
+                {"logPrefix", LogPrefix()});
 
             Status = EStatus::Free;
             AfterRelease();
@@ -251,7 +267,8 @@ bool TPartitionFamily::Reset(ETargetStatus targetStatus, const TActorContext& ct
 
             auto it = Consumer.Families.find(MergeTo);
             if (it == Consumer.Families.end()) {
-                PQ_LOG_D(" has been released for merge but target family is not exists.");
+                YDB_LOG_DEBUG("Has been released for merge but target family is not exists",
+                    {"logPrefix", LogPrefix()});
                 return true;
             }
             auto* targetFamily = it->second.get();
@@ -266,7 +283,8 @@ bool TPartitionFamily::Reset(ETargetStatus targetStatus, const TActorContext& ct
 }
 
 void TPartitionFamily::Destroy(const TActorContext&) {
-    PQ_LOG_D(" destroyed.");
+    YDB_LOG_DEBUG("Destroyed",
+        {"logPrefix", LogPrefix()});
 
     if (Session) {
         Session->Families.erase(Id);
@@ -301,11 +319,14 @@ void TPartitionFamily::AfterRelease() {
 
 void TPartitionFamily::StartReading(TSession& session, const TActorContext& ctx) {
     if (Status != EStatus::Free) {
-        PQ_LOG_CRIT("try start reading but the family status is " << Status);
+        YDB_LOG_CRIT("Try start reading but the family status is",
+            {"logPrefix", LogPrefix()},
+            {"status", Status});
         return;
     }
 
-    PQ_LOG_T("start reading");
+    YDB_LOG_TRACE("Start reading",
+        {"logPrefix", LogPrefix()});
 
     Status = EStatus::Active;
 
@@ -327,7 +348,9 @@ void TPartitionFamily::StartReading(TSession& session, const TActorContext& ctx)
 }
 
 void TPartitionFamily::AttachePartitions(const std::vector<ui32>& partitions, const TActorContext& ctx) {
-    PQ_LOG_D("attaching partitions [" << JoinRange(", ", partitions.begin(), partitions.end()) << "]");
+    YDB_LOG_DEBUG("Attaching partitions",
+        {"logPrefix", LogPrefix()},
+        {"partitions", JoinRange(", ", partitions.begin(), partitions.end())});
 
     std::unordered_set<ui32> existedPartitions;
     existedPartitions.insert(Partitions.begin(), Partitions.end());
@@ -376,13 +399,17 @@ void TPartitionFamily::AttachePartitions(const std::vector<ui32>& partitions, co
 }
 
 void TPartitionFamily::ActivatePartition(ui32 partitionId) {
-    PQ_LOG_D("activating partition " << partitionId);
+    YDB_LOG_DEBUG("Activating partition",
+        {"logPrefix", LogPrefix()},
+        {"partitionId", partitionId});
 
     ChangePartitionCounters(1, -1);
 }
 
 void TPartitionFamily::InactivatePartition(ui32 partitionId) {
-    PQ_LOG_D("inactivating partition " << partitionId);
+    YDB_LOG_DEBUG("Inactivating partition",
+        {"logPrefix", LogPrefix()},
+        {"partitionId", partitionId});
 
     ChangePartitionCounters(-1, 1);
 }
@@ -401,7 +428,9 @@ void TPartitionFamily::InactivatePartition(ui32 partitionId) {
  }
 
 void TPartitionFamily::Merge(TPartitionFamily* other) {
-    PQ_LOG_D("merge family with  " << other->DebugStr());
+    YDB_LOG_DEBUG("Merge family with",
+        {"logPrefix", LogPrefix()},
+        {"debug", other->DebugStr()});
 
     Y_VERIFY(this != other);
 
@@ -545,8 +574,12 @@ void TPartitionFamily::UpdateSpecialSessions() {
 void TPartitionFamily::LockPartition(ui32 partitionId, const TActorContext& ctx) {
     auto step = NextStep();
 
-    PQ_LOG_I("lock partition " << partitionId << " for " << Session->DebugStr()
-            << " generation " << TabletGeneration() << " step " << step);
+    YDB_LOG_INFO("Lock partition for generation step",
+        {"logPrefix", LogPrefix()},
+        {"partitionId", partitionId},
+        {"debug", Session->DebugStr()},
+        {"tabletGeneration", TabletGeneration()},
+        {"step", step});
 
     ctx.Send(Session->Sender, MakeEvLockPartition(partitionId, step).release());
 }
@@ -636,7 +669,9 @@ ui32 TConsumer::NextStep() {
 void TConsumer::RegisterPartition(ui32 partitionId, const TActorContext& ctx) {
     auto [_, inserted] = Partitions.try_emplace(partitionId, TPartition());
     if (inserted && IsReadable(partitionId)) {
-        PQ_LOG_D("register readable partition " << partitionId);
+        YDB_LOG_DEBUG("Register readable partition",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId});
 
         CreateFamily({partitionId}, ctx);
     }
@@ -666,7 +701,9 @@ TPartitionFamily* TConsumer::CreateFamily(std::vector<ui32>&& partitions, TParti
         UnreadableFamilies[id] = family;
     }
 
-    PQ_LOG_D("family created " << family->DebugStr());
+    YDB_LOG_DEBUG("Family created",
+        {"logPrefix", LogPrefix()},
+        {"family", family->DebugStr()});
 
     return family;
 }
@@ -706,7 +743,10 @@ bool TConsumer::BreakUpFamily(TPartitionFamily* family, ui32 partitionId, bool d
     std::vector<TPartitionFamily*> newFamilies;
 
     if (!family->IsLonely()) {
-        PQ_LOG_D("break up " << family->DebugStr() << " partition=" << partitionId);
+        YDB_LOG_DEBUG("Break up",
+            {"logPrefix", LogPrefix()},
+            {"family", family->DebugStr()},
+            {"partition", partitionId});
 
         std::unordered_set<ui32> partitions;
         partitions.insert(family->Partitions.begin(), family->Partitions.end());
@@ -786,7 +826,10 @@ bool TConsumer::BreakUpFamily(TPartitionFamily* family, ui32 partitionId, bool d
                 }
             }
         } else {
-            PQ_LOG_D("can't break up " << family->DebugStr() << " because partition=" << partitionId << " is not root of family");
+            YDB_LOG_DEBUG("Can't break up because is not root of family",
+                {"logPrefix", LogPrefix()},
+                {"family", family->DebugStr()},
+                {"partition", partitionId});
         }
     }
 
@@ -868,7 +911,9 @@ TPartitionFamily* TConsumer::FindFamily(ui32 partitionId) {
 }
 
 void TConsumer::RegisterReadingSession(TSession* session, const TActorContext& ctx) {
-    PQ_LOG_I("register reading session " << session->DebugStr());
+    YDB_LOG_INFO("Register reading session",
+        {"logPrefix", LogPrefix()},
+        {"debug", session->DebugStr()});
 
     Sessions[session->Pipe] = session;
 
@@ -952,7 +997,9 @@ void TConsumer::UnregisterReadingSession(TSession* session, const TActorContext&
 bool TConsumer::Unlock(const TActorId& sender, ui32 partitionId, const TActorContext& ctx) {
     auto* family = FindFamily(partitionId);
     if (!family) {
-        PQ_LOG_CRIT("unlocking the partition " << partitionId << " from unknown family.");
+        YDB_LOG_CRIT("Unlocking the partition from unknown family",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId});
         return false;
     }
 
@@ -1034,7 +1081,10 @@ bool TConsumer::ProccessReadingFinished(ui32 partitionId, bool wasInactive, cons
     });
 
     if (partition.NeedReleaseChildren()) {
-        PQ_LOG_D("Attache partitions [" << JoinRange(", ", newPartitions.begin(), newPartitions.end()) << "] to " << family->DebugStr());
+        YDB_LOG_DEBUG("Attache partitions",
+            {"logPrefix", LogPrefix()},
+            {"newPartitions", JoinRange(", ", newPartitions.begin(), newPartitions.end())},
+            {"family", family->DebugStr()});
         for (auto id : newPartitions) {
             if (family->CanAttach(std::vector{id})) {
                 auto* node = GetPartitionGraph().GetPartition(id);
@@ -1066,7 +1116,10 @@ bool TConsumer::ProccessReadingFinished(ui32 partitionId, bool wasInactive, cons
                     }
                 }
             } else {
-                PQ_LOG_D("Can't attache partition " << id << " to " << family->DebugStr());
+                YDB_LOG_DEBUG("Can't attache partition",
+                    {"logPrefix", LogPrefix()},
+                    {"id", id},
+                    {"family", family->DebugStr()});
             }
         }
     } else {
@@ -1083,19 +1136,28 @@ bool TConsumer::ProccessReadingFinished(ui32 partitionId, bool wasInactive, cons
 
 void TConsumer::StartReading(ui32 partitionId, const TActorContext& ctx) {
     if (!GetPartitionInfo(partitionId)) {
-        PQ_LOG_NOTICE("Reading of the partition " << partitionId << " was started by " << ConsumerName << " but partition has been deleted.");
+        YDB_LOG_NOTICE("Reading of the partition was started by but partition has been deleted",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
         return;
     }
 
     auto* partition = GetPartition(partitionId);
     if (!partition) {
-        PQ_LOG_NOTICE("Reading of the partition " << partitionId << " was started by " << ConsumerName << " but partition does not exist.");
+        YDB_LOG_NOTICE("Reading of the partition was started by but partition does not exist",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
         return;
     }
 
     auto wasInactive = partition->IsInactive();
     if (partition->StartReading()) {
-        PQ_LOG_D("Reading of the partition " << partitionId << " was started by " << ConsumerName << ". We stop reading from child partitions.");
+        YDB_LOG_DEBUG("Reading of the partition was started by We stop reading from child partitions",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
 
         auto* family = FindFamily(partitionId);
         if (!family) {
@@ -1137,21 +1199,27 @@ void TConsumer::FinishReading(TEvPersQueue::TEvReadingPartitionFinishedRequest::
     auto partitionId = r.GetPartitionId();
 
     if (!IsReadable(partitionId)) {
-        PQ_LOG_D("Reading of the partition " << partitionId << " was finished by " << ConsumerName
-                << " but the partition isn't readable");
+        YDB_LOG_DEBUG("Reading of the partition was finished by but the partition isn't readable",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
         return;
     }
 
     auto* family = FindFamily(partitionId);
     if (!family) {
-        PQ_LOG_D("Reading of the partition " << partitionId << " was finished by " << ConsumerName
-                << " but the partition hasn't family");
+        YDB_LOG_DEBUG("Reading of the partition was finished by but the partition hasn't family",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
         return;
     }
 
     if (!family->Session) {
-        PQ_LOG_D("Reading of the partition " << partitionId << " was finished by " << ConsumerName
-                << " but the partition hasn't reading session");
+        YDB_LOG_DEBUG("Reading of the partition was finished by but the partition hasn't reading session",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", ConsumerName});
         return;
     }
 
@@ -1159,8 +1227,12 @@ void TConsumer::FinishReading(TEvPersQueue::TEvReadingPartitionFinishedRequest::
 
     const bool wasInactive = partition.IsInactive();
     if (partition.SetFinishedState(r.GetScaleAwareSDK(), r.GetStartedReadingFromEndOffset()) || wasInactive) {
-        PQ_LOG_D("Reading of the partition " << partitionId << " was finished by " << r.GetConsumer()
-                << ", firstMessage=" << r.GetStartedReadingFromEndOffset() << ", " << GetSdkDebugString0(r.GetScaleAwareSDK()));
+        YDB_LOG_DEBUG("Reading of the partition was finished by",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumer", r.GetConsumer()},
+            {"firstMessage", r.GetStartedReadingFromEndOffset()},
+            {"scaleAwareSdk", GetSdkDebugString0(r.GetScaleAwareSDK())});
 
         if (ProccessReadingFinished(partitionId, wasInactive, ctx)) {
             ScheduleBalance(ctx);
@@ -1168,9 +1240,13 @@ void TConsumer::FinishReading(TEvPersQueue::TEvReadingPartitionFinishedRequest::
     } else if (!partition.IsInactive()) {
         auto delay = std::min<size_t>(1ul << partition.Iteration, Balancer.GetLifetimeSeconds()); // TODO use split/merge time
 
-        PQ_LOG_D("Reading of the partition " << partitionId << " was finished by " << r.GetConsumer()
-                << ". Scheduled release of the partition for re-reading. Delay=" << delay << " seconds,"
-                << " firstMessage=" << r.GetStartedReadingFromEndOffset() << ", " << GetSdkDebugString0(r.GetScaleAwareSDK()));
+        YDB_LOG_DEBUG("Reading of the partition was finished by Scheduled release of the partition for re-reading. seconds",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumer", r.GetConsumer()},
+            {"delay", delay},
+            {"firstMessage", r.GetStartedReadingFromEndOffset()},
+            {"scaleAwareSdk", GetSdkDebugString0(r.GetScaleAwareSDK())});
 
         ctx.Schedule(TDuration::Seconds(delay), new TEvPQ::TEvWakeupReleasePartition(ConsumerName, partitionId, partition.Cookie));
     }
@@ -1178,13 +1254,15 @@ void TConsumer::FinishReading(TEvPersQueue::TEvReadingPartitionFinishedRequest::
 
 void TConsumer::ScheduleBalance(const TActorContext& ctx) {
     if (BalanceScheduled) {
-        PQ_LOG_T("rebalancing already was scheduled");
+        YDB_LOG_TRACE("Rebalancing already was scheduled",
+            {"logPrefix", LogPrefix()});
         return;
     }
 
     BalanceScheduled = true;
 
-    PQ_LOG_D("rebalancing was scheduled");
+    YDB_LOG_DEBUG("Rebalancing was scheduled",
+        {"logPrefix", LogPrefix()});
 
     ctx.Send(Balancer.TopicActor.SelfId(), new TEvPQ::TEvBalanceConsumer(ConsumerName));
 }
@@ -1254,9 +1332,14 @@ size_t GetMaxFamilySize(const std::unordered_map<size_t, const std::unique_ptr<T
 }
 
 void TConsumer::Balance(const TActorContext& ctx) {
-    PQ_LOG_D("balancing. Sessions=" << Sessions.size() << ", Families=" << Families.size()
-            << ", UnreadableFamilies=" << UnreadableFamilies.size() << " [" << DebugStr(UnreadableFamilies)
-            << "], RequireBalancing=" << FamiliesRequireBalancing.size() << " [" << DebugStr(FamiliesRequireBalancing) << "]");
+    YDB_LOG_DEBUG("Balancing",
+        {"logPrefix", LogPrefix()},
+        {"sessions", Sessions.size()},
+        {"families", Families.size()},
+        {"unreadableFamilies", UnreadableFamilies.size()},
+        {"unreadableFamiliesDebug", DebugStr(UnreadableFamilies)},
+        {"requireBalancing", FamiliesRequireBalancing.size()},
+        {"familiesRequireBalancingDebug", DebugStr(FamiliesRequireBalancing)});
 
     if (Sessions.empty()) {
         return;
@@ -1270,7 +1353,9 @@ void TConsumer::Balance(const TActorContext& ctx) {
             continue;
         }
         if (!family->SpecialSessions.contains(family->Session->Pipe)) {
-            PQ_LOG_D("rebalance " << family->DebugStr() << " because exists the special session for it");
+            YDB_LOG_DEBUG("Rebalance because exists the special session for it",
+                {"logPrefix", LogPrefix()},
+                {"family", family->DebugStr()});
             family->Release(ctx);
         }
     }
@@ -1293,7 +1378,9 @@ void TConsumer::Balance(const TActorContext& ctx) {
             }
 
             if (sit == sessions.end()) {
-                PQ_LOG_D("balancing of the " << family->DebugStr() << " failed because there are no suitable reading sessions.");
+                YDB_LOG_DEBUG("Balancing of the failed because there are no suitable reading sessions",
+                    {"logPrefix", LogPrefix()},
+                    {"family", family->DebugStr()});
 
                 continue;
             }
@@ -1303,7 +1390,10 @@ void TConsumer::Balance(const TActorContext& ctx) {
             // Reorder sessions
             sessions.erase(sit);
 
-            PQ_LOG_D("balancing " << family->DebugStr() << " for " << session->DebugStr());
+            YDB_LOG_DEBUG("Balancing",
+                {"logPrefix", LogPrefix()},
+                {"family", family->DebugStr()},
+                {"debug", session->DebugStr()});
             family->StartReading(*session, ctx);
 
             // Reorder sessions
@@ -1322,8 +1412,12 @@ void TConsumer::Balance(const TActorContext& ctx) {
         auto desiredFamilyCount = familyCount / commonSessions.size();
         auto allowPlusOne = familyCount % commonSessions.size();
 
-        PQ_LOG_D("start rebalancing. familyCount=" << familyCount << ", sessionCount=" << commonSessions.size()
-                << ", desiredFamilyCount=" << desiredFamilyCount << ", allowPlusOne=" << allowPlusOne);
+        YDB_LOG_DEBUG("Start rebalancing",
+            {"logPrefix", LogPrefix()},
+            {"familyCount", familyCount},
+            {"sessionCount", commonSessions.size()},
+            {"desiredFamilyCount", desiredFamilyCount},
+            {"allowPlusOne", allowPlusOne});
 
         if (!OrderedSessions) {
             OrderedSessions.emplace();
@@ -1352,7 +1446,9 @@ void TConsumer::Balance(const TActorContext& ctx) {
             auto* family = it->second;
 
             if (!family->IsActive()) {
-                PQ_LOG_D("skip balancing " << family->DebugStr() << " because it is not active.");
+                YDB_LOG_DEBUG("Skip balancing because it is not active",
+                    {"logPrefix", LogPrefix()},
+                    {"family", family->DebugStr()});
 
                 it = FamiliesRequireBalancing.erase(it);
                 continue;
@@ -1365,14 +1461,18 @@ void TConsumer::Balance(const TActorContext& ctx) {
             }
 
             if (family->Session->ActiveFamilyCount == 1) {
-                PQ_LOG_D("skip balancing " << family->DebugStr() << " because it is considered a session that does not read anything else.");
+                YDB_LOG_DEBUG("Skip balancing because it is considered a session that does not read anything else",
+                    {"logPrefix", LogPrefix()},
+                    {"family", family->DebugStr()});
 
                 it = FamiliesRequireBalancing.erase(it);
                 continue;
             }
 
             if (family->SpecialSessions.size() <= 1) {
-                PQ_LOG_D("skip balancing " << family->DebugStr() << " because there are no other suitable reading sessions.");
+                YDB_LOG_DEBUG("Skip balancing because there are no other suitable reading sessions",
+                    {"logPrefix", LogPrefix()},
+                    {"family", family->DebugStr()});
 
                 it = FamiliesRequireBalancing.erase(it);
                 continue;
@@ -1394,14 +1494,18 @@ void TConsumer::Balance(const TActorContext& ctx) {
                 family->Release(ctx);
                 it = FamiliesRequireBalancing.erase(it);
             } else {
-                PQ_LOG_D("skip balancing " << family->DebugStr() << " because it is already being read by the best session.");
+                YDB_LOG_DEBUG("Skip balancing because it is already being read by the best session",
+                    {"logPrefix", LogPrefix()},
+                    {"family", family->DebugStr()});
                 ++it;
             }
         }
     }
 
     auto duration = TInstant::Now() - startTime;
-    PQ_LOG_D("balancing duration: " << duration);
+    YDB_LOG_DEBUG("Balancing",
+        {"logPrefix", LogPrefix()},
+        {"duration", duration});
 }
 
 void TConsumer::Release(ui32 partitionId, const TActorContext& ctx) {
@@ -1517,8 +1621,10 @@ const std::unordered_map<TActorId, std::unique_ptr<TSession>>& TBalancer::GetSes
 
 
 void TBalancer::UpdateConfig(const std::vector<ui32>& addedPartitions, const std::vector<ui32>& deletedPartitions, const TActorContext& ctx) {
-    PQ_LOG_D("updating configuration. Deleted partitions [" << JoinRange(", ", deletedPartitions.begin(), deletedPartitions.end())
-            << "]. Added partitions [" << JoinRange(", ", addedPartitions.begin(), addedPartitions.end()) << "]");
+    YDB_LOG_DEBUG("Updating configuration. Deleted partitions Added partitions",
+        {"logPrefix", LogPrefix()},
+        {"deletedPartitions", JoinRange(", ", deletedPartitions.begin(), deletedPartitions.end())},
+        {"addedPartitions", JoinRange(", ", addedPartitions.begin(), addedPartitions.end())});
 
     for (auto partitionId : deletedPartitions) {
         for (auto& [_, consumer] : Consumers) {
@@ -1544,14 +1650,19 @@ bool TBalancer::SetCommittedState(const TString& consumerName, ui32 partitionId,
     }
 
     if (!consumer->IsReadable(partitionId)) {
-        PQ_LOG_D("The offset of the partition " << partitionId << " was commited by " << consumerName
-                << " but the partition isn't readable");
+        YDB_LOG_DEBUG("The offset of the partition was commited by but the partition isn't readable",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", consumerName});
         return false;
     }
 
     auto wasInactive = consumer->IsInactive(partitionId);
     if (consumer->SetCommittedState(partitionId, generation, cookie)) {
-        PQ_LOG_D("The offset of the partition " << partitionId << " was commited by " << consumerName);
+        YDB_LOG_DEBUG("The offset of the partition was commited by",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", partitionId},
+            {"consumerName", consumerName});
 
         if (consumer->ProccessReadingFinished(partitionId, wasInactive, ctx)) {
             consumer->ScheduleBalance(ctx);
@@ -1575,13 +1686,17 @@ void TBalancer::Handle(TEvPersQueue::TEvReadingPartitionStartedRequest::TPtr& ev
     auto pipeClient = ActorIdFromProto(r.GetPipeClient());
 
     if (pipeClient && !Sessions.contains(pipeClient)) {
-        PQ_LOG_D("Received TEvReadingPartitionStartedRequest from unknown pipe " << pipeClient);
+        YDB_LOG_DEBUG("Received TEvReadingPartitionStartedRequest from unknown pipe",
+            {"logPrefix", LogPrefix()},
+            {"pipeClient", pipeClient});
         return;
     }
 
     auto consumer = GetConsumer(r.GetConsumer());
     if (!consumer) {
-        PQ_LOG_D("Received TEvReadingPartitionStartedRequest from unknown consumer " << r.GetConsumer());
+        YDB_LOG_DEBUG("Received TEvReadingPartitionStartedRequest from unknown consumer",
+            {"logPrefix", LogPrefix()},
+            {"consumer", r.GetConsumer()});
         return;
     }
 
@@ -1593,13 +1708,17 @@ void TBalancer::Handle(TEvPersQueue::TEvReadingPartitionFinishedRequest::TPtr& e
     auto pipeClient = ActorIdFromProto(r.GetPipeClient());
 
     if (pipeClient && !Sessions.contains(pipeClient)) {
-        PQ_LOG_D("Received TEvReadingPartitionFinishedRequest from unknown pipe " << pipeClient);
+        YDB_LOG_DEBUG("Received TEvReadingPartitionFinishedRequest from unknown pipe",
+            {"logPrefix", LogPrefix()},
+            {"pipeClient", pipeClient});
         return;
     }
 
     auto consumer = GetConsumer(r.GetConsumer());
     if (!consumer) {
-        PQ_LOG_D("Received TEvReadingPartitionFinishedRequest from unknown consumer " << r.GetConsumer());
+        YDB_LOG_DEBUG("Received TEvReadingPartitionFinishedRequest from unknown consumer",
+            {"logPrefix", LogPrefix()},
+            {"consumer", r.GetConsumer()});
         return;
     }
 
@@ -1614,17 +1733,28 @@ void TBalancer::Handle(TEvPersQueue::TEvPartitionReleased::TPtr& ev, const TActo
 
     auto* partitionInfo = GetPartitionInfo(partitionId);
     if (!partitionInfo) {
-        PQ_LOG_CRIT("client " << r.GetClientId() << " pipe " << sender << " got deleted partition " << r);
+        YDB_LOG_CRIT("Client pipe got deleted partition",
+            {"logPrefix", LogPrefix()},
+            {"clientId", r.GetClientId()},
+            {"sender", sender},
+            {"r", r});
         return;
     }
 
-    PQ_LOG_I("client " << r.GetClientId() << " released partition from pipe " << sender
-            << " session " << r.GetSession() << " partition " << partitionId);
+    YDB_LOG_INFO("Client released partition from pipe session partition",
+        {"logPrefix", LogPrefix()},
+        {"clientId", r.GetClientId()},
+        {"sender", sender},
+        {"session", r.GetSession()},
+        {"partitionId", partitionId});
 
     auto* consumer = GetConsumer(consumerName);
     if (!consumer) {
-        PQ_LOG_CRIT("client " << r.GetClientId() << " pipe " << sender
-                << " is not connected and got release partitions request for session " << r.GetSession());
+        YDB_LOG_CRIT("Client pipe is not connected and got release partitions request for session",
+            {"logPrefix", LogPrefix()},
+            {"clientId", r.GetClientId()},
+            {"sender", sender},
+            {"session", r.GetSession()});
         return;
     }
 
@@ -1646,11 +1776,17 @@ void TBalancer::Handle(TEvPQ::TEvWakeupReleasePartition::TPtr &ev, const TActorC
     }
 
     if (partition->Commited) {
-        PQ_LOG_D("skip releasing partition " << msg->PartitionId << " of consumer \"" << msg->Consumer << "\" by reading finished timeout because offset is commited");
+        YDB_LOG_DEBUG("Skip releasing partition of consumer by reading finished timeout because offset is commited",
+            {"logPrefix", LogPrefix()},
+            {"partitionId", msg->PartitionId},
+            {"consumer", msg->Consumer});
         return;
     }
 
-    PQ_LOG_I("releasing partition " << msg->PartitionId << " of consumer \"" << msg->Consumer << "\" by reading finished timeout");
+    YDB_LOG_INFO("Releasing partition of consumer by reading finished timeout",
+        {"logPrefix", LogPrefix()},
+        {"partitionId", msg->PartitionId},
+        {"consumer", msg->Consumer});
 
     consumer->Release(msg->PartitionId, ctx);
 }
@@ -1666,22 +1802,31 @@ void TBalancer::Handle(TEvTabletPipe::TEvServerConnected::TPtr& ev, const TActor
     auto& session = it->second;
     ++session->ServerActors;
 
-    PQ_LOG_I("pipe " << sender << " connected; active server actors: " << session->ServerActors);
+    YDB_LOG_INFO("Pipe connected; active server",
+        {"logPrefix", LogPrefix()},
+        {"sender", sender},
+        {"actors", session->ServerActors});
 }
 
 void TBalancer::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr& ev, const TActorContext& ctx) {
-    PQ_LOG_D("pipe " << ev->Get()->ClientId << " disconnected.");
+    YDB_LOG_DEBUG("Pipe disconnected",
+        {"logPrefix", LogPrefix()},
+        {"clientId", ev->Get()->ClientId});
     Subscriptions.erase(ev->Get()->ClientId);
 
     auto it = Sessions.find(ev->Get()->ClientId);
 
     if (it == Sessions.end()) {
-        PQ_LOG_D("pipe " << ev->Get()->ClientId << " disconnected but there aren't sessions exists.");
+        YDB_LOG_DEBUG("Pipe disconnected but there aren't sessions exists",
+            {"logPrefix", LogPrefix()},
+            {"clientId", ev->Get()->ClientId});
         return;
     }
 
-    PQ_LOG_I("pipe " << ev->Get()->ClientId << " disconnected; active server actors: "
-            << (it != Sessions.end() ? it->second->ServerActors : -1));
+    YDB_LOG_INFO("Pipe disconnected; active server",
+        {"logPrefix", LogPrefix()},
+        {"clientId", ev->Get()->ClientId},
+        {"actors", (it != Sessions.end() ? it->second->ServerActors : -1)});
 
     auto& session = it->second;
     if (--(session->ServerActors) > 0) {
@@ -1689,8 +1834,11 @@ void TBalancer::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr& ev, const TAc
     }
 
     if (!session->SessionName.empty()) {
-        PQ_LOG_NOTICE("pipe " << ev->Get()->ClientId << " client "
-                << session->ClientId << " disconnected session " << session->SessionName);
+        YDB_LOG_NOTICE("Pipe client disconnected session",
+            {"logPrefix", LogPrefix()},
+            {"eventClientId", ev->Get()->ClientId},
+            {"sessionClientId", session->ClientId},
+            {"sessionName", session->SessionName});
 
         auto* consumer = GetConsumer(session->ClientId);
         if (consumer) {
@@ -1706,7 +1854,9 @@ void TBalancer::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr& ev, const TAc
 
         Sessions.erase(it);
     } else {
-        PQ_LOG_I("pipe " << ev->Get()->ClientId << " disconnected no session");
+        YDB_LOG_INFO("Pipe disconnected no session",
+            {"logPrefix", LogPrefix()},
+            {"clientId", ev->Get()->ClientId});
 
         Sessions.erase(it);
     }
@@ -1717,27 +1867,37 @@ void TBalancer::Handle(TEvPersQueue::TEvRegisterReadSession::TPtr& ev, const TAc
     auto& consumerName = r.GetClientId();
 
     TActorId pipe = ActorIdFromProto(r.GetPipeClient());
-    PQ_LOG_NOTICE("consumer \"" << consumerName << "\" register session for pipe " << pipe << " session " << r.GetSession());
+    YDB_LOG_NOTICE("Consumer register session for pipe session",
+        {"logPrefix", LogPrefix()},
+        {"consumerName", consumerName},
+        {"pipe", pipe},
+        {"session", r.GetSession()});
 
     if (consumerName.empty()) {
-        PQ_LOG_CRIT("ignored the session registration with empty consumer name.");
+        YDB_LOG_CRIT("Ignored the session registration with empty consumer name",
+            {"logPrefix", LogPrefix()});
         return;
     }
 
     if (r.GetSession().empty()) {
-        PQ_LOG_CRIT("ignored the session registration with empty session name.");
+        YDB_LOG_CRIT("Ignored the session registration with empty session name",
+            {"logPrefix", LogPrefix()});
         return;
     }
 
     if (!pipe) {
-        PQ_LOG_CRIT("ignored the session registration with empty Pipe.");
+        YDB_LOG_CRIT("Ignored the session registration with empty Pipe",
+            {"logPrefix", LogPrefix()});
         return;
     }
 
     auto jt = Sessions.find(pipe);
     if (jt == Sessions.end()) {
-        PQ_LOG_CRIT("client \"" << consumerName << "\" pipe " << pipe
-                        << " is not connected and got register session request for session " << r.GetSession());
+        YDB_LOG_CRIT("Client pipe is not connected and got register session request for session",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName},
+            {"pipe", pipe},
+            {"session", r.GetSession()});
         return;
     }
 
@@ -1852,7 +2012,9 @@ void TBalancer::Handle(TEvPersQueue::TEvStatusResponse::TPtr& ev, const TActorCo
 }
 
 void TBalancer::ProcessPendingStats(const TActorContext& ctx) {
-    PQ_LOG_D("ProcessPendingStats. PendingUpdates size " << PendingUpdates.size());
+    YDB_LOG_DEBUG("ProcessPendingStats. PendingUpdates size",
+        {"logPrefix", LogPrefix()},
+        {"pendingUpdatesSize", PendingUpdates.size()});
 
     GetPartitionGraph().Travers([&](ui32 id) {
         for (auto& d : PendingUpdates[id]) {
@@ -1868,7 +2030,9 @@ void TBalancer::ProcessPendingStats(const TActorContext& ctx) {
 
 void TBalancer::Handle(TEvPersQueue::TEvBalancingSubscribe::TPtr& ev, const TActorContext& ctx) {
     auto& record = ev->Get()->Record;
-    PQ_LOG_D("Handle TEvPersQueue::TEvBalancingSubscribe " << record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPersQueue::TEvBalancingSubscribe",
+        {"logPrefix", LogPrefix()},
+        {"ev", record.ShortDebugString()});
 
     auto sender = ActorIdFromProto(record.GetSourceActor());
     auto status = Consumers.contains(record.GetConsumer()) ?
@@ -1880,7 +2044,9 @@ void TBalancer::Handle(TEvPersQueue::TEvBalancingSubscribe::TPtr& ev, const TAct
 
 void TBalancer::Handle(TEvPersQueue::TEvBalancingUnsubscribe::TPtr& ev, const TActorContext&) {
     auto& record = ev->Get()->Record;
-    PQ_LOG_D("Handle TEvPersQueue::TEvBalancingUnsubscribe " << record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPersQueue::TEvBalancingUnsubscribe",
+        {"logPrefix", LogPrefix()},
+        {"ev", record.ShortDebugString()});
 
     auto sender = ActorIdFromProto(record.GetSourceActor());
     auto& consumer = record.GetConsumer();
diff --git a/ydb/core/persqueue/pqrb/read_balancer__balancing_app.cpp b/ydb/core/persqueue/pqrb/read_balancer__balancing_app.cpp
index d62bd55c870..370de6c38d0 100644
--- a/ydb/core/persqueue/pqrb/read_balancer__balancing_app.cpp
+++ b/ydb/core/persqueue/pqrb/read_balancer__balancing_app.cpp
@@ -3,9 +3,6 @@
 #include <library/cpp/monlib/service/pages/templates.h>
 #include <ydb/core/persqueue/common/common_app.h>
 
-#define DEBUG(message)
-
-
 namespace NKikimr::NPQ::NBalancing {
 
 void TBalancer::RenderApp(NApp::TNavigationBar& __navigationBar) const {
diff --git a/ydb/core/persqueue/pqrb/read_balancer__mlp_balancing.cpp b/ydb/core/persqueue/pqrb/read_balancer__mlp_balancing.cpp
index 61030f78738..2fdf743951d 100644
--- a/ydb/core/persqueue/pqrb/read_balancer__mlp_balancing.cpp
+++ b/ydb/core/persqueue/pqrb/read_balancer__mlp_balancing.cpp
@@ -3,6 +3,8 @@
 
 #include <util/generic/ymath.h>
 
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
+
 namespace NKikimr::NPQ::NBalancing {
 
 namespace {
@@ -192,7 +194,9 @@ void TMLPConsumer::Rebuild() {
         }
     }
 
-    PQ_LOG_D("Rebuild " << JoinSeq(",", PartitionsForBalancing) << " partitions for balancing");
+    YDB_LOG_DEBUG("Rebuild partitions for balancing",
+        {"logPrefix", LogPrefix()},
+        {"partitionsForBalancing", JoinSeq(",", PartitionsForBalancing)});
 }
 
 const TMLPConsumer::TMetrics& TMLPConsumer::GetMetrics() const {
@@ -212,7 +216,9 @@ TPrepareGetPartitionResponse TMLPBalancer::PrepareGetPartitionResponse(
 
     auto* consumerConfig = NPQ::GetConsumer(GetConfig(), consumerName);
     if (!consumerConfig) {
-        PQ_LOG_D("Consumer '" << consumerName << "' does not exist");
+        YDB_LOG_DEBUG("Consumer does not exist",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         result.IsError = true;
         result.ErrorStatus = Ydb::StatusIds::SCHEME_ERROR;
         result.ErrorMessage = TStringBuilder() << "Consumer '" << consumerName << "' does not exist";
@@ -220,7 +226,9 @@ TPrepareGetPartitionResponse TMLPBalancer::PrepareGetPartitionResponse(
     }
 
     if (consumerConfig->GetType() != NKikimrPQ::TPQTabletConfig::CONSUMER_TYPE_MLP) {
-        PQ_LOG_D("Consumer '" << consumerName << "' is not MLP consumer");
+        YDB_LOG_DEBUG("Consumer is not MLP consumer",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         result.IsError = true;
         result.ErrorStatus = Ydb::StatusIds::SCHEME_ERROR;
         result.ErrorMessage = TStringBuilder() << "Consumer '" << consumerName << "' is not MLP consumer";
@@ -270,14 +278,18 @@ void TMLPBalancer::Handle(TEvPQ::TEvMLPGetRuntimeAttributesRequest::TPtr& ev) {
 
     const auto* consumerConfig = NPQ::GetConsumer(GetConfig(), consumerName);
     if (!consumerConfig) {
-        PQ_LOG_D("Consumer '" << consumerName << "' does not exist");
+        YDB_LOG_DEBUG("Consumer does not exist",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         TopicActor.Send(ev->Sender, new TEvPQ::TEvMLPErrorResponse(Ydb::StatusIds::SCHEME_ERROR,
             TStringBuilder() << "Consumer '" << consumerName << "' does not exist"), 0, ev->Cookie);
         return;
     }
 
     if (consumerConfig->GetType() != NKikimrPQ::TPQTabletConfig::CONSUMER_TYPE_MLP) {
-        PQ_LOG_D("Consumer '" << consumerName << "' is not MLP consumer");
+        YDB_LOG_DEBUG("Consumer is not MLP consumer",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         TopicActor.Send(ev->Sender, new TEvPQ::TEvMLPErrorResponse(Ydb::StatusIds::SCHEME_ERROR,
             TStringBuilder() << "Consumer '" << consumerName << "' is not MLP consumer"), 0, ev->Cookie);
         return;
@@ -285,7 +297,9 @@ void TMLPBalancer::Handle(TEvPQ::TEvMLPGetRuntimeAttributesRequest::TPtr& ev) {
 
     auto it = Consumers.find(consumerName);
     if (it == Consumers.end()) {
-        PQ_LOG_D("Consumer '" << consumerName << "' is not initialized");
+        YDB_LOG_DEBUG("Consumer is not initialized",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         TopicActor.Send(ev->Sender, new TEvPQ::TEvMLPGetRuntimeAttributesResponse(0, 0, 0), 0, ev->Cookie);
         return;
     }
@@ -302,7 +316,9 @@ void TMLPBalancer::Handle(TEvPQ::TEvMLPGetRuntimeAttributesRequest::TPtr& ev) {
 }
 
 void TMLPBalancer::Handle(TEvPersQueue::TEvStatusResponse::TPtr& ev, const TActorContext&) {
-    PQ_LOG_D("Handle TEvPersQueue::TEvStatusResponse " << ev->Get()->Record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPersQueue::TEvStatusResponse",
+        {"logPrefix", LogPrefix()},
+        {"ev", ev->Get()->Record.ShortDebugString()});
 
     absl::flat_hash_map<TString, bool> mlpConsumers;
     for (const auto& consumer : GetConfig().GetConsumers()) {
@@ -347,7 +363,9 @@ void TMLPBalancer::Handle(TEvPersQueue::TEvStatusResponse::TPtr& ev, const TActo
 
 void TMLPBalancer::Handle(TEvPQ::TEvReadingPartitionStatusRequest::TPtr& ev, const TActorContext&) {
     auto& record = ev->Get()->Record;
-    PQ_LOG_D("Handle TEvPQ::TEvReadingPartitionStatusRequest " << record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPQ::TEvReadingPartitionStatusRequest",
+        {"logPrefix", LogPrefix()},
+        {"ev", record.ShortDebugString()});
     SetUseForReading(record.GetConsumer(),
                      record.GetPartitionId(),
                      true, // reading is finished
@@ -363,7 +381,9 @@ void TMLPBalancer::Handle(TEvPQ::TEvReadingPartitionStatusRequest::TPtr& ev, con
 
 void TMLPBalancer::Handle(TEvPQ::TEvMLPConsumerStatus::TPtr& ev) {
     auto& record = ev->Get()->Record;
-    PQ_LOG_D("Handle TEvPQ::TEvMLPConsumerStatus " << record.ShortDebugString());
+    YDB_LOG_DEBUG("Handle TEvPQ::TEvMLPConsumerStatus",
+        {"logPrefix", LogPrefix()},
+        {"ev", record.ShortDebugString()});
     SetUseForReading(record.GetConsumer(),
                      record.GetPartitionId(),
                      std::nullopt, // reading is finished
@@ -418,12 +438,16 @@ void TMLPBalancer::SetUseForReading(const TString& consumerName,
                                     ui64 cookie) {
     auto* consumerConfig = NPQ::GetConsumer(GetConfig(), consumerName);
     if (!consumerConfig) {
-        PQ_LOG_D("Consumer '" << consumerName << "' does not exist");
+        YDB_LOG_DEBUG("Consumer does not exist",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         return;
     }
 
     if (consumerConfig->GetType() != NKikimrPQ::TPQTabletConfig::CONSUMER_TYPE_MLP) {
-        PQ_LOG_D("Consumer '" << consumerName << "' is not MLP consumer");
+        YDB_LOG_DEBUG("Consumer is not MLP consumer",
+            {"logPrefix", LogPrefix()},
+            {"consumerName", consumerName});
         return;
     }
 
diff --git a/ydb/core/persqueue/pqrb/read_balancer_log.h b/ydb/core/persqueue/pqrb/read_balancer_log.h
index 96890d1d810..934eb09106e 100644
--- a/ydb/core/persqueue/pqrb/read_balancer_log.h
+++ b/ydb/core/persqueue/pqrb/read_balancer_log.h
@@ -7,12 +7,5 @@ namespace NKikimr::NPQ {
 
 inline TString LogPrefix() { return {}; }
 
-#define PQ_LOG_T(stream) LOG_TRACE_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_D(stream) LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_I(stream) LOG_INFO_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_W(stream) LOG_WARN_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_NOTICE(stream) LOG_NOTICE_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_ERROR(stream) LOG_ERROR_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
-#define PQ_LOG_CRIT(stream) LOG_CRIT_S(*NActors::TlsActivationContext, NKikimrServices::PERSQUEUE_READ_BALANCER, LogPrefix() << stream)
 
 } // namespace NKikimr::NPQ
diff --git a/ydb/core/persqueue/public/schema/schema_operation.cpp b/ydb/core/persqueue/public/schema/schema_operation.cpp
index 0f98bb13e42..99ab505f3cf 100644
--- a/ydb/core/persqueue/public/schema/schema_operation.cpp
+++ b/ydb/core/persqueue/public/schema/schema_operation.cpp
@@ -177,10 +177,9 @@ private:
     }
 
     void ReplyErrorAndDie(Ydb::StatusIds::StatusCode errorCode, TString&& errorMessage) {
-        YDB_LOG_DEBUG("Dump NPQLOGPREFIX, replyErrorAndDie, errorMessage",
+        YDB_LOG_DEBUG(errorMessage,
             {"logPrefix", NPQ_LOG_PREFIX},
-            {"replyErrorAndDie", errorCode},
-            {"errorMessage", errorMessage});
+            {"replyErrorAndDie", errorCode});
         Send(ParentId, new TEvSchemaOperationResponse(errorCode, std::move(errorMessage)), 0, Cookie);
         PassAway();
     }
diff --git a/ydb/core/persqueue/writer/writer.cpp b/ydb/core/persqueue/writer/writer.cpp
index f1ca11529d9..9a68f23d15a 100644
--- a/ydb/core/persqueue/writer/writer.cpp
+++ b/ydb/core/persqueue/writer/writer.cpp
@@ -318,7 +318,7 @@ class TPartitionWriter : public TActorBootstrapped<TPartitionWriter>, public TPa
 
         WriteId = NPQ::GetWriteId(record.GetResponse().GetTopicOperations());
 
-        YDB_LOG_DEBUG("Dump LOGPREFIX, sessionId, txId, writeId",
+        YDB_LOG_DEBUG("",
             {"logPrefix", LOG_PREFIX},
             {"sessionId", Opts.SessionId},
             {"txId", Opts.TxId},