[backups] refactor change sender (#5354)

18 files changed, 942 insertions, 882 deletions

diff --git a/ydb/core/change_exchange/change_exchange.cpp b/ydb/core/change_exchange/change_exchange.cpp
index 92df9b3cea7..dfb46f1173b 100644
--- a/ydb/core/change_exchange/change_exchange.cpp
+++ b/ydb/core/change_exchange/change_exchange.cpp
@@ -88,21 +88,26 @@ TString TEvChangeExchange::TEvRemoveRecords::ToString() const {
     << " }";
 }
 
-/// TEvRecords
-TEvChangeExchange::TEvRecords::TEvRecords(const TVector<IChangeRecord::TPtr>& records)
+// TEvRecords
+TEvChangeExchange::TEvRecords::TEvRecords(const TChangeRecordVector& records)
     : Records(records)
 {
 }
 
-TEvChangeExchange::TEvRecords::TEvRecords(TVector<IChangeRecord::TPtr>&& records)
+TEvChangeExchange::TEvRecords::TEvRecords(TChangeRecordVector&& records)
     : Records(std::move(records))
 {
 }
 
+
 TString TEvChangeExchange::TEvRecords::ToString() const {
-    return TStringBuilder() << ToStringHeader() << " {"
-        << " Records [" << JoinSeq(",", Records) << "]"
-    << " }";
+    return std::visit(
+        [&](auto& records){
+            return TStringBuilder() << ToStringHeader() << " {"
+                << " Records " << ((TBaseChangeRecordContainer*)records.get())->Out()
+            << " }";
+        },
+        Records);
 }
 
 /// TEvForgetRecords
diff --git a/ydb/core/change_exchange/change_exchange.h b/ydb/core/change_exchange/change_exchange.h
index 9cd000b4c23..2b42e9b90f9 100644
--- a/ydb/core/change_exchange/change_exchange.h
+++ b/ydb/core/change_exchange/change_exchange.h
@@ -8,8 +8,33 @@
 
 #include <util/generic/vector.h>
 
+namespace NKikimr {
+
+namespace NDataShard {
+    class TChangeRecord;
+}
+
+namespace NReplication::NService {
+    class TChangeRecord;
+}
+
+struct TBaseChangeRecordContainer {
+    virtual ~TBaseChangeRecordContainer() = default;
+    virtual TString Out() = 0;
+};
+
+template <class T>
+struct TChangeRecordContainer {};
+
+}
+
 namespace NKikimr::NChangeExchange {
 
+using TChangeRecordVector = std::variant<
+    std::shared_ptr<TChangeRecordContainer<NDataShard::TChangeRecord>>,
+    std::shared_ptr<TChangeRecordContainer<NReplication::NService::TChangeRecord>>
+>;
+
 struct TEvChangeExchange {
     enum EEv {
         // Enqueue for sending
@@ -73,10 +98,10 @@ struct TEvChangeExchange {
     };
 
     struct TEvRecords: public TEventLocal<TEvRecords, EvRecords> {
-        TVector<IChangeRecord::TPtr> Records;
+        TChangeRecordVector Records;
 
-        explicit TEvRecords(const TVector<IChangeRecord::TPtr>& records);
-        explicit TEvRecords(TVector<IChangeRecord::TPtr>&& records);
+        explicit TEvRecords(const TChangeRecordVector& records);
+        explicit TEvRecords(TChangeRecordVector&& records);
         TString ToString() const override;
     };
 
diff --git a/ydb/core/change_exchange/change_record.h b/ydb/core/change_exchange/change_record.h
index 88959dd0dc7..e47175afa20 100644
--- a/ydb/core/change_exchange/change_record.h
+++ b/ydb/core/change_exchange/change_record.h
@@ -6,6 +6,8 @@
 
 namespace NKikimr::NChangeExchange {
 
+class IChangeSenderResolver;
+
 class IChangeRecord: public TThrRefBase {
 public:
     using TPtr = TIntrusivePtr<IChangeRecord>;
@@ -34,11 +36,7 @@ public:
     virtual TString ToString() const = 0;
     virtual void Out(IOutputStream& out) const = 0;
 
-    template <typename T>
-    T* Get() {
-        return dynamic_cast<T*>(this);
-    }
-
+    virtual ui64 ResolvePartitionId(IChangeSenderResolver* const resolver) const = 0;
 }; // IChangeRecord
 
 template <typename T, typename TDerived> class TChangeRecordBuilder;
@@ -77,13 +75,11 @@ protected:
 public:
     TChangeRecordBuilder()
         : Record(MakeIntrusive<T>())
-    {
-    }
+    {}
 
-    explicit TChangeRecordBuilder(IChangeRecord::TPtr record)
+    explicit TChangeRecordBuilder(TIntrusivePtr<T> record)
         : Record(std::move(record))
-    {
-    }
+    {}
 
     TSelf& WithOrder(ui64 order) {
         GetRecord()->Order = order;
@@ -105,12 +101,12 @@ public:
         return static_cast<TSelf&>(*this);
     }
 
-    IChangeRecord::TPtr Build() {
+    TIntrusivePtr<T> Build() {
         return Record;
     }
 
 protected:
-    IChangeRecord::TPtr Record;
+    TIntrusivePtr<T> Record;
 
 }; // TChangeRecordBuilder
 
@@ -119,3 +115,7 @@ protected:
 Y_DECLARE_OUT_SPEC(inline, NKikimr::NChangeExchange::IChangeRecord::TPtr, out, value) {
     return value->Out(out);
 }
+
+Y_DECLARE_OUT_SPEC(inline, NKikimr::NChangeExchange::IChangeRecord*, out, value) {
+    return value->Out(out);
+}
diff --git a/ydb/core/change_exchange/change_sender_common_ops.cpp b/ydb/core/change_exchange/change_sender_common_ops.cpp
deleted file mode 100644
index 8b10355730a..00000000000
--- a/ydb/core/change_exchange/change_sender_common_ops.cpp
+++ /dev/null
@@ -1,641 +0,0 @@
-#include "change_sender_common_ops.h"
-#include "change_sender_monitoring.h"
-
-#include <ydb/library/yverify_stream/yverify_stream.h>
-
-#include <library/cpp/monlib/service/pages/mon_page.h>
-#include <library/cpp/monlib/service/pages/templates.h>
-
-#include <util/generic/algorithm.h>
-#include <util/generic/size_literals.h>
-
-namespace NKikimr::NChangeExchange {
-
-void TBaseChangeSender::LazyCreateSender(THashMap<ui64, TSender>& senders, ui64 partitionId) {
-    auto res = senders.emplace(partitionId, TSender{});
-    Y_ABORT_UNLESS(res.second);
-
-    for (const auto& [order, broadcast] : Broadcasting) {
-        if (AddBroadcastPartition(order, partitionId)) {
-            // re-enqueue record to send it in the correct order
-            Enqueued.insert(ReEnqueue(broadcast.Record));
-        }
-    }
-}
-
-void TBaseChangeSender::RegisterSender(ui64 partitionId) {
-    Y_ABORT_UNLESS(Senders.contains(partitionId));
-    auto& sender = Senders.at(partitionId);
-
-    Y_ABORT_UNLESS(!sender.ActorId);
-    sender.ActorId = ActorOps->RegisterWithSameMailbox(CreateSender(partitionId));
-}
-
-void TBaseChangeSender::CreateMissingSenders(const TVector<ui64>& partitionIds) {
-    THashMap<ui64, TSender> senders;
-
-    for (const auto& partitionId : partitionIds) {
-        auto it = Senders.find(partitionId);
-        if (it != Senders.end()) {
-            senders.emplace(partitionId, std::move(it->second));
-            Senders.erase(it);
-        } else {
-            LazyCreateSender(senders, partitionId);
-        }
-    }
-
-    for (const auto& [partitionId, sender] : Senders) {
-        ReEnqueueRecords(sender);
-        ProcessBroadcasting(&TBaseChangeSender::RemoveBroadcastPartition,
-            partitionId, sender.Broadcasting);
-        if (sender.ActorId) {
-            ActorOps->Send(sender.ActorId, new TEvents::TEvPoisonPill());
-        }
-    }
-
-    Senders = std::move(senders);
-}
-
-void TBaseChangeSender::RecreateSenders(const TVector<ui64>& partitionIds) {
-    for (const auto& partitionId : partitionIds) {
-        LazyCreateSender(Senders, partitionId);
-    }
-}
-
-void TBaseChangeSender::CreateSenders(const TVector<ui64>& partitionIds, bool partitioningChanged) {
-    if (partitioningChanged) {
-        CreateMissingSenders(partitionIds);
-    } else {
-        RecreateSenders(GonePartitions);
-    }
-
-    GonePartitions.clear();
-
-    if (!Enqueued || !RequestRecords()) {
-        SendRecords();
-    }
-}
-
-void TBaseChangeSender::KillSenders() {
-    for (const auto& [_, sender] : std::exchange(Senders, {})) {
-        if (sender.ActorId) {
-            ActorOps->Send(sender.ActorId, new TEvents::TEvPoisonPill());
-        }
-    }
-}
-
-void TBaseChangeSender::EnqueueRecords(TVector<TEvChangeExchange::TEvEnqueueRecords::TRecordInfo>&& records) {
-    for (auto& record : records) {
-        Y_VERIFY_S(PathId == record.PathId, "Unexpected record's path id"
-            << ": expected# " << PathId
-            << ", got# " << record.PathId);
-        Enqueued.emplace(record.Order, record.BodySize);
-    }
-
-    RequestRecords();
-}
-
-bool TBaseChangeSender::RequestRecords() {
-    if (!Enqueued) {
-        return false;
-    }
-
-    auto it = Enqueued.begin();
-    TVector<TIncompleteRecord> records;
-
-    bool exceeded = false;
-    while (it != Enqueued.end()) {
-        if (MemUsage && (MemUsage + it->BodySize) > MemLimit) {
-            if (!it->ReEnqueued || exceeded) {
-                break;
-            }
-
-            exceeded = true;
-        }
-
-        MemUsage += it->BodySize;
-
-        records.emplace_back(it->Order, it->BodySize);
-        PendingBody.emplace(it->Order, it->BodySize);
-        it = Enqueued.erase(it);
-    }
-
-    if (!records) {
-        return false;
-    }
-
-    ActorOps->Send(GetChangeServer(), new TEvChangeExchange::TEvRequestRecords(std::move(records)));
-    return true;
-}
-
-void TBaseChangeSender::ProcessRecords(TVector<IChangeRecord::TPtr>&& records) {
-    for (auto& record : records) {
-        auto it = PendingBody.find(record->GetOrder());
-        if (it == PendingBody.end()) {
-            continue;
-        }
-
-        if (it->BodySize != record->GetBody().size()) {
-            MemUsage -= it->BodySize;
-            MemUsage += record->GetBody().size();
-        }
-
-        if (record->IsBroadcast()) {
-            // assume that broadcast records are too small to affect memory consumption
-            MemUsage -= record->GetBody().size();
-        }
-
-        PendingSent.emplace(record->GetOrder(), std::move(record));
-        PendingBody.erase(it);
-    }
-
-    SendRecords();
-}
-
-void TBaseChangeSender::SendRecords() {
-    if (!Resolver->IsResolved()) {
-        return;
-    }
-
-    if (!PendingSent) {
-        return;
-    }
-
-    auto it = PendingSent.begin();
-    THashSet<ui64> sendTo;
-    THashSet<ui64> registrations;
-    bool needToResolve = false;
-
-    while (it != PendingSent.end()) {
-        if (Enqueued && Enqueued.begin()->Order <= it->first) {
-            break;
-        }
-
-        if (PendingBody && PendingBody.begin()->Order <= it->first) {
-            break;
-        }
-
-        if (!it->second->IsBroadcast()) {
-            const ui64 partitionId = Resolver->GetPartitionId(it->second);
-            if (!Senders.contains(partitionId)) {
-                needToResolve = true;
-                ++it;
-                continue;
-            }
-
-            auto& sender = Senders.at(partitionId);
-            sender.Prepared.push_back(std::move(it->second));
-            if (!sender.ActorId) {
-                Y_ABORT_UNLESS(!sender.Ready);
-                registrations.insert(partitionId);
-            }
-            if (sender.Ready) {
-                sendTo.insert(partitionId);
-            }
-        } else {
-            auto& broadcast = EnsureBroadcast(it->second);
-            EraseNodesIf(broadcast.PendingPartitions, [&](ui64 partitionId) {
-                if (Senders.contains(partitionId)) {
-                    auto& sender = Senders.at(partitionId);
-                    sender.Prepared.push_back(std::move(it->second));
-                    if (!sender.ActorId) {
-                        Y_ABORT_UNLESS(!sender.Ready);
-                        registrations.insert(partitionId);
-                    }
-                    if (sender.Ready) {
-                        sendTo.insert(partitionId);
-                    }
-
-                    return true;
-                }
-
-                return false;
-            });
-        }
-
-        it = PendingSent.erase(it);
-    }
-
-    for (const auto partitionId : registrations) {
-        RegisterSender(partitionId);
-    }
-
-    for (const auto partitionId : sendTo) {
-        SendPreparedRecords(partitionId);
-    }
-
-    if (needToResolve && !Resolver->IsResolving()) {
-        Resolver->Resolve();
-    }
-
-    RequestRecords();
-}
-
-void TBaseChangeSender::ForgetRecords(TVector<ui64>&& records) {
-    for (const auto& record : records) {
-        auto it = PendingBody.find(record);
-        if (it == PendingBody.end()) {
-            continue;
-        }
-
-        MemUsage -= it->BodySize;
-        PendingBody.erase(it);
-    }
-
-    RequestRecords();
-}
-
-void TBaseChangeSender::OnReady(ui64 partitionId) {
-    auto it = Senders.find(partitionId);
-    if (it == Senders.end()) {
-        return;
-    }
-
-    auto& sender = it->second;
-    sender.Ready = true;
-
-    if (sender.Pending) {
-        RemoveRecords(std::exchange(sender.Pending, {}));
-    }
-
-    if (sender.Broadcasting) {
-        ProcessBroadcasting(&TBaseChangeSender::CompleteBroadcastPartition,
-            partitionId, std::exchange(sender.Broadcasting, {}));
-    }
-
-    if (sender.Prepared) {
-        SendPreparedRecords(partitionId);
-    }
-
-    RequestRecords();
-}
-
-void TBaseChangeSender::OnGone(ui64 partitionId) {
-    auto it = Senders.find(partitionId);
-    if (it == Senders.end()) {
-        return;
-    }
-
-    ReEnqueueRecords(it->second);
-    Senders.erase(it);
-    GonePartitions.push_back(partitionId);
-
-    if (Resolver->IsResolving()) {
-        return;
-    }
-
-    Resolver->Resolve();
-}
-
-void TBaseChangeSender::SendPreparedRecords(ui64 partitionId) {
-    Y_ABORT_UNLESS(Senders.contains(partitionId));
-    auto& sender = Senders.at(partitionId);
-
-    Y_ABORT_UNLESS(sender.Ready);
-    sender.Ready = false;
-
-    sender.Pending.reserve(sender.Prepared.size());
-    for (const auto& record : sender.Prepared) {
-        if (!record->IsBroadcast()) {
-            sender.Pending.emplace_back(record->GetOrder(), record->GetBody().size());
-            MemUsage -= record->GetBody().size();
-        } else {
-            sender.Broadcasting.push_back(record->GetOrder());
-        }
-    }
-
-    Y_ABORT_UNLESS(sender.ActorId);
-    ActorOps->Send(sender.ActorId, new TEvChangeExchange::TEvRecords(std::exchange(sender.Prepared, {})));
-}
-
-void TBaseChangeSender::ReEnqueueRecords(const TSender& sender) {
-    for (const auto& record : sender.Pending) {
-        Enqueued.insert(ReEnqueue(record));
-    }
-
-    for (const auto& record : sender.Prepared) {
-        if (!record->IsBroadcast()) {
-            Enqueued.insert(ReEnqueue(record->GetOrder(), record->GetBody().size()));
-            MemUsage -= record->GetBody().size();
-        }
-    }
-}
-
-TBaseChangeSender::TBroadcast& TBaseChangeSender::EnsureBroadcast(IChangeRecord::TPtr record) {
-    Y_ABORT_UNLESS(record->IsBroadcast());
-
-    auto it = Broadcasting.find(record->GetOrder());
-    if (it != Broadcasting.end()) {
-        return it->second;
-    }
-
-    THashSet<ui64> partitionIds;
-    for (const auto& [partitionId, _] : Senders) {
-        partitionIds.insert(partitionId);
-    }
-    for (const auto partitionId : GonePartitions) {
-        partitionIds.insert(partitionId);
-    }
-
-    auto res = Broadcasting.emplace(record->GetOrder(), TBroadcast{
-        .Record = {record->GetOrder(), record->GetBody().size()},
-        .Partitions = partitionIds,
-        .PendingPartitions = partitionIds,
-    });
-
-    return res.first->second;
-}
-
-bool TBaseChangeSender::AddBroadcastPartition(ui64 order, ui64 partitionId) {
-    auto it = Broadcasting.find(order);
-    Y_ABORT_UNLESS(it != Broadcasting.end());
-
-    auto& broadcast = it->second;
-    if (broadcast.Partitions.contains(partitionId)) {
-        return false;
-    }
-
-    broadcast.Partitions.insert(partitionId);
-    broadcast.PendingPartitions.insert(partitionId);
-
-    return true;
-}
-
-bool TBaseChangeSender::RemoveBroadcastPartition(ui64 order, ui64 partitionId) {
-    auto it = Broadcasting.find(order);
-    Y_ABORT_UNLESS(it != Broadcasting.end());
-
-    auto& broadcast = it->second;
-    broadcast.Partitions.erase(partitionId);
-    broadcast.PendingPartitions.erase(partitionId);
-    broadcast.CompletedPartitions.erase(partitionId);
-
-    return MaybeCompleteBroadcast(order);
-}
-
-bool TBaseChangeSender::CompleteBroadcastPartition(ui64 order, ui64 partitionId) {
-    auto it = Broadcasting.find(order);
-    Y_ABORT_UNLESS(it != Broadcasting.end());
-
-    auto& broadcast = it->second;
-    broadcast.CompletedPartitions.insert(partitionId);
-
-    return MaybeCompleteBroadcast(order);
-}
-
-bool TBaseChangeSender::MaybeCompleteBroadcast(ui64 order) {
-    auto it = Broadcasting.find(order);
-    Y_ABORT_UNLESS(it != Broadcasting.end());
-
-    auto& broadcast = it->second;
-    if (broadcast.PendingPartitions || broadcast.Partitions.size() != broadcast.CompletedPartitions.size()) {
-        return false;
-    }
-
-    Broadcasting.erase(it);
-    return true;
-}
-
-void TBaseChangeSender::ProcessBroadcasting(std::function<bool(TBaseChangeSender*, ui64, ui64)> f,
-        ui64 partitionId, const TVector<ui64>& broadcasting)
-{
-    TVector<ui64> remove;
-    for (const auto order : broadcasting) {
-        if (std::invoke(f, this, order, partitionId)) {
-            remove.push_back(order);
-        }
-    }
-
-    if (remove) {
-        RemoveRecords(std::move(remove));
-    }
-}
-
-void TBaseChangeSender::RemoveRecords() {
-    THashSet<ui64> remove;
-
-    for (const auto& record : std::exchange(Enqueued, {})) {
-        remove.insert(record.Order);
-    }
-
-    for (const auto& record : std::exchange(PendingBody, {})) {
-        remove.insert(record.Order);
-    }
-
-    for (const auto& [order, _] : std::exchange(PendingSent, {})) {
-        remove.insert(order);
-    }
-
-    for (const auto& [order, _] : std::exchange(Broadcasting, {})) {
-        remove.insert(order);
-    }
-
-    for (const auto& [_, sender] : Senders) {
-        for (const auto& record : sender.Pending) {
-            remove.insert(record.Order);
-        }
-        for (const auto& record : sender.Prepared) {
-            remove.insert(record->GetOrder());
-        }
-    }
-
-    if (remove) {
-        RemoveRecords(TVector<ui64>(remove.begin(), remove.end()));
-    }
-}
-
-TBaseChangeSender::TBaseChangeSender(IActorOps* actorOps, IChangeSenderResolver* resolver, const TPathId& pathId)
-    : ActorOps(actorOps)
-    , Resolver(resolver)
-    , PathId(pathId)
-    , MemLimit(192_KB)
-    , MemUsage(0)
-{
-}
-
-void TBaseChangeSender::RenderHtmlPage(ui64 tabletId, NMon::TEvRemoteHttpInfo::TPtr& ev,
-        const TActorContext& ctx)
-{
-    const auto& cgi = ev->Get()->Cgi();
-    if (const auto& str = cgi.Get("partitionId")) {
-        ui64 partitionId = 0;
-        if (TryFromString(str, partitionId)) {
-            auto it = Senders.find(partitionId);
-            if (it != Senders.end()) {
-                if (const auto& to = it->second.ActorId) {
-                    ctx.Send(ev->Forward(to));
-                } else {
-                    ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes(TStringBuilder()
-                        << "Change sender '" << PathId << ":" << partitionId << "' is not running"));
-                }
-            } else {
-                ActorOps->Send(ev->Sender, new NMon::TEvRemoteBinaryInfoRes(NMonitoring::HTTPNOTFOUND));
-            }
-        } else {
-            ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes("Invalid partitionId"));
-        }
-
-        return;
-    }
-
-    TStringStream html;
-
-    HTML(html) {
-        Header(html, "Change sender", tabletId);
-
-        SimplePanel(html, "Info", [this](IOutputStream& html) {
-            HTML(html) {
-                DL_CLASS("dl-horizontal") {
-                    TermDesc(html, "MemLimit", MemLimit);
-                    TermDesc(html, "MemUsage", MemUsage);
-                }
-            }
-        });
-
-        SimplePanel(html, "Partition senders", [this, tabletId](IOutputStream& html) {
-            HTML(html) {
-                TABLE_CLASS("table table-hover") {
-                    TABLEHEAD() {
-                        TABLER() {
-                            TABLEH() { html << "#"; }
-                            TABLEH() { html << "PartitionId"; }
-                            TABLEH() { html << "Ready"; }
-                            TABLEH() { html << "Pending"; }
-                            TABLEH() { html << "Prepared"; }
-                            TABLEH() { html << "Broadcasting"; }
-                            TABLEH() { html << "Actor"; }
-                        }
-                    }
-                    TABLEBODY() {
-                        ui32 i = 0;
-                        for (const auto& [partitionId, sender] : Senders) {
-                            TABLER() {
-                                TABLED() { html << ++i; }
-                                TABLED() { html << partitionId; }
-                                TABLED() { html << sender.Ready; }
-                                TABLED() { html << sender.Pending.size(); }
-                                TABLED() { html << sender.Prepared.size(); }
-                                TABLED() { html << sender.Broadcasting.size(); }
-                                TABLED() { ActorLink(html, tabletId, PathId, partitionId); }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-
-        CollapsedPanel(html, "Enqueued", "enqueued", [this](IOutputStream& html) {
-            HTML(html) {
-                TABLE_CLASS("table table-hover") {
-                    TABLEHEAD() {
-                        TABLER() {
-                            TABLEH() { html << "#"; }
-                            TABLEH() { html << "Order"; }
-                            TABLEH() { html << "BodySize"; }
-                        }
-                    }
-                    TABLEBODY() {
-                        ui32 i = 0;
-                        for (const auto& record : Enqueued) {
-                            TABLER() {
-                                TABLED() { html << ++i; }
-                                TABLED() { html << record.Order; }
-                                TABLED() { html << record.BodySize; }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-
-        CollapsedPanel(html, "PendingBody", "pendingBody", [this](IOutputStream& html) {
-            HTML(html) {
-                TABLE_CLASS("table table-hover") {
-                    TABLEHEAD() {
-                        TABLER() {
-                            TABLEH() { html << "#"; }
-                            TABLEH() { html << "Order"; }
-                            TABLEH() { html << "BodySize"; }
-                        }
-                    }
-                    TABLEBODY() {
-                        ui32 i = 0;
-                        for (const auto& record : PendingBody) {
-                            TABLER() {
-                                TABLED() { html << ++i; }
-                                TABLED() { html << record.Order; }
-                                TABLED() { html << record.BodySize; }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-
-        CollapsedPanel(html, "PendingSent", "pendingSent", [this](IOutputStream& html) {
-            HTML(html) {
-                TABLE_CLASS("table table-hover") {
-                    TABLEHEAD() {
-                        TABLER() {
-                            TABLEH() { html << "#"; }
-                            TABLEH() { html << "Order"; }
-                            TABLEH() { html << "Group"; }
-                            TABLEH() { html << "Step"; }
-                            TABLEH() { html << "TxId"; }
-                            TABLEH() { html << "Kind"; }
-                            TABLEH() { html << "Source"; }
-                        }
-                    }
-                    TABLEBODY() {
-                        ui32 i = 0;
-                        for (const auto& [order, record] : PendingSent) {
-                            TABLER() {
-                                TABLED() { html << ++i; }
-                                TABLED() { html << order; }
-                                TABLED() { html << record->GetGroup(); }
-                                TABLED() { html << record->GetStep(); }
-                                TABLED() { html << record->GetTxId(); }
-                                TABLED() { html << record->GetKind(); }
-                                TABLED() { html << record->GetSource(); }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-
-        CollapsedPanel(html, "Broadcasting", "broadcasting", [this](IOutputStream& html) {
-            HTML(html) {
-                TABLE_CLASS("table table-hover") {
-                    TABLEHEAD() {
-                        TABLER() {
-                            TABLEH() { html << "#"; }
-                            TABLEH() { html << "Order"; }
-                            TABLEH() { html << "BodySize"; }
-                            TABLEH() { html << "Partitions"; }
-                            TABLEH() { html << "PendingPartitions"; }
-                            TABLEH() { html << "CompletedPartitions"; }
-                        }
-                    }
-                    TABLEBODY() {
-                        ui32 i = 0;
-                        for (const auto& [order, broadcast] : Broadcasting) {
-                            TABLER() {
-                                TABLED() { html << ++i; }
-                                TABLED() { html << order; }
-                                TABLED() { html << broadcast.Record.BodySize; }
-                                TABLED() { html << broadcast.Partitions.size(); }
-                                TABLED() { html << broadcast.PendingPartitions.size(); }
-                                TABLED() { html << broadcast.CompletedPartitions.size(); }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-    }
-
-    ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes(html.Str()));
-}
-
-}
diff --git a/ydb/core/change_exchange/change_sender_common_ops.h b/ydb/core/change_exchange/change_sender_common_ops.h
index 504afbde9da..0b684b30705 100644
--- a/ydb/core/change_exchange/change_sender_common_ops.h
+++ b/ydb/core/change_exchange/change_sender_common_ops.h
@@ -1,15 +1,24 @@
 #pragma once
 
 #include "change_exchange.h"
+#include "change_sender_resolver.h"
+
+#include <ydb/core/change_exchange/change_sender_monitoring.h>
 
 #include <ydb/library/actors/core/actor.h>
 #include <ydb/library/actors/core/mon.h>
+#include <ydb/library/yverify_stream/yverify_stream.h>
+
+#include <library/cpp/monlib/service/pages/mon_page.h>
+#include <library/cpp/monlib/service/pages/templates.h>
 
 #include <util/generic/hash.h>
 #include <util/generic/map.h>
 #include <util/generic/set.h>
 #include <util/string/builder.h>
 
+#include <concepts>
+
 namespace NKikimr::NChangeExchange {
 
 struct TEvChangeExchangePrivate {
@@ -56,36 +65,17 @@ struct TEvChangeExchangePrivate {
 
 }; // TEvChangeExchangePrivate
 
-class IChangeSender {
+class ISenderFactory {
 public:
-    virtual ~IChangeSender() = default;
-
-    virtual TActorId GetChangeServer() const = 0;
-
-    virtual void CreateSenders(const TVector<ui64>& partitionIds, bool partitioningChanged = true) = 0;
-    virtual void KillSenders() = 0;
-    virtual IActor* CreateSender(ui64 partitionId) = 0;
-    virtual void RemoveRecords() = 0;
-
-    virtual void EnqueueRecords(TVector<TEvChangeExchange::TEvEnqueueRecords::TRecordInfo>&& records) = 0;
-    virtual void ProcessRecords(TVector<IChangeRecord::TPtr>&& records) = 0;
-    virtual void ForgetRecords(TVector<ui64>&& records) = 0;
-    virtual void OnReady(ui64 partitionId) = 0;
-    virtual void OnGone(ui64 partitionId) = 0;
+    virtual ~ISenderFactory() = default;
+    virtual IActor* CreateSender(ui64 partitionId) const = 0;
 };
 
-class IChangeSenderResolver {
-public:
-    virtual ~IChangeSenderResolver() = default;
-
-    virtual void Resolve() = 0;
-    virtual bool IsResolving() const = 0;
-    virtual bool IsResolved() const = 0;
-    virtual ui64 GetPartitionId(IChangeRecord::TPtr record) const = 0;
-};
-
-class TBaseChangeSender: public IChangeSender {
+template <class TChangeRecord>
+class TBaseChangeSender {
     using TIncompleteRecord = TEvChangeExchange::TEvRequestRecords::TRecordInfo;
+    // we need this to safely cast and call Out on a container
+    static_assert(std::derived_from<TChangeRecordContainer<TChangeRecord>, TBaseChangeRecordContainer>);
 
     struct TEnqueuedRecord: TIncompleteRecord {
         bool ReEnqueued = false;
@@ -108,7 +98,7 @@ class TBaseChangeSender: public IChangeSender {
         TActorId ActorId;
         bool Ready = false;
         TVector<TIncompleteRecord> Pending;
-        TVector<IChangeRecord::TPtr> Prepared;
+        TVector<typename TChangeRecord::TPtr> Prepared;
         TVector<ui64> Broadcasting;
     };
 
@@ -119,24 +109,292 @@ class TBaseChangeSender: public IChangeSender {
         THashSet<ui64> CompletedPartitions;
     };
 
-    void LazyCreateSender(THashMap<ui64, TSender>& senders, ui64 partitionId);
-    void RegisterSender(ui64 partitionId);
-    void CreateMissingSenders(const TVector<ui64>& partitionIds);
-    void RecreateSenders(const TVector<ui64>& partitionIds);
+    void LazyCreateSender(THashMap<ui64, TSender>& senders, ui64 partitionId) {
+        auto res = senders.emplace(partitionId, TSender{});
+        Y_ABORT_UNLESS(res.second);
+
+        for (const auto& [order, broadcast] : Broadcasting) {
+            if (AddBroadcastPartition(order, partitionId)) {
+                // re-enqueue record to send it in the correct order
+                Enqueued.insert(ReEnqueue(broadcast.Record));
+            }
+        }
+    }
+
+    void RegisterSender(ui64 partitionId) {
+        Y_ABORT_UNLESS(Senders.contains(partitionId));
+        auto& sender = Senders.at(partitionId);
+
+        Y_ABORT_UNLESS(!sender.ActorId);
+        sender.ActorId = ActorOps->RegisterWithSameMailbox(SenderFactory->CreateSender(partitionId));
+    }
+
+    void CreateMissingSenders(const TVector<ui64>& partitionIds) {
+        THashMap<ui64, TSender> senders;
+
+        for (const auto& partitionId : partitionIds) {
+            auto it = Senders.find(partitionId);
+            if (it != Senders.end()) {
+                senders.emplace(partitionId, std::move(it->second));
+                Senders.erase(it);
+            } else {
+                LazyCreateSender(senders, partitionId);
+            }
+        }
+
+        for (const auto& [partitionId, sender] : Senders) {
+            ReEnqueueRecords(sender);
+            ProcessBroadcasting(&TBaseChangeSender::RemoveBroadcastPartition,
+                partitionId, sender.Broadcasting);
+            if (sender.ActorId) {
+                ActorOps->Send(sender.ActorId, new TEvents::TEvPoisonPill());
+            }
+        }
+
+        Senders = std::move(senders);
+    }
+
+    void RecreateSenders(const TVector<ui64>& partitionIds) {
+        for (const auto& partitionId : partitionIds) {
+            LazyCreateSender(Senders, partitionId);
+        }
+    }
+
+    bool RequestRecords() {
+        if (!Enqueued) {
+            return false;
+        }
+
+        auto it = Enqueued.begin();
+        TVector<TIncompleteRecord> records;
+
+        bool exceeded = false;
+        while (it != Enqueued.end()) {
+            if (MemUsage && (MemUsage + it->BodySize) > MemLimit) {
+                if (!it->ReEnqueued || exceeded) {
+                    break;
+                }
+
+                exceeded = true;
+            }
+
+            MemUsage += it->BodySize;
+
+            records.emplace_back(it->Order, it->BodySize);
+            PendingBody.emplace(it->Order, it->BodySize);
+            it = Enqueued.erase(it);
+        }
+
+        if (!records) {
+            return false;
+        }
+
+        ActorOps->Send(GetChangeServer(), new TEvChangeExchange::TEvRequestRecords(std::move(records)));
+        return true;
+    }
+
+    void SendRecords() {
+        if (!Resolver->IsResolved()) {
+            return;
+        }
+
+        if (!PendingSent) {
+            return;
+        }
+
+        auto it = PendingSent.begin();
+        THashSet<ui64> sendTo;
+        THashSet<ui64> registrations;
+        bool needToResolve = false;
+
+        while (it != PendingSent.end()) {
+            if (Enqueued && Enqueued.begin()->Order <= it->first) {
+                break;
+            }
+
+            if (PendingBody && PendingBody.begin()->Order <= it->first) {
+                break;
+            }
+
+            if (!it->second->IsBroadcast()) {
+                const ui64 partitionId = it->second->ResolvePartitionId(Resolver);
+                if (!Senders.contains(partitionId)) {
+                    needToResolve = true;
+                    ++it;
+                    continue;
+                }
+
+                auto& sender = Senders.at(partitionId);
+                sender.Prepared.push_back(std::move(it->second));
+                if (!sender.ActorId) {
+                    Y_ABORT_UNLESS(!sender.Ready);
+                    registrations.insert(partitionId);
+                }
+                if (sender.Ready) {
+                    sendTo.insert(partitionId);
+                }
+            } else {
+                auto& broadcast = EnsureBroadcast(it->second);
+                EraseNodesIf(broadcast.PendingPartitions, [&](ui64 partitionId) {
+                    if (Senders.contains(partitionId)) {
+                        auto& sender = Senders.at(partitionId);
+                        sender.Prepared.push_back(std::move(it->second));
+                        if (!sender.ActorId) {
+                            Y_ABORT_UNLESS(!sender.Ready);
+                            registrations.insert(partitionId);
+                        }
+                        if (sender.Ready) {
+                            sendTo.insert(partitionId);
+                        }
+
+                        return true;
+                    }
+
+                    return false;
+                });
+            }
+
+            it = PendingSent.erase(it);
+        }
+
+        for (const auto partitionId : registrations) {
+            RegisterSender(partitionId);
+        }
+
+        for (const auto partitionId : sendTo) {
+            SendPreparedRecords(partitionId);
+        }
+
+        if (needToResolve && !Resolver->IsResolving()) {
+            Resolver->Resolve();
+        }
+
+        RequestRecords();
+    }
+
+    void SendPreparedRecords(ui64 partitionId) {
+        Y_ABORT_UNLESS(Senders.contains(partitionId));
+        auto& sender = Senders.at(partitionId);
+
+        Y_ABORT_UNLESS(sender.Ready);
+        sender.Ready = false;
+
+        sender.Pending.reserve(sender.Prepared.size());
+        for (const auto& record : sender.Prepared) {
+            if (!record->IsBroadcast()) {
+                sender.Pending.emplace_back(record->GetOrder(), record->GetBody().size());
+                MemUsage -= record->GetBody().size();
+            } else {
+                sender.Broadcasting.push_back(record->GetOrder());
+            }
+        }
+
+        Y_ABORT_UNLESS(sender.ActorId);
+        ActorOps->Send(sender.ActorId, new TEvChangeExchange::TEvRecords(std::make_shared<TChangeRecordContainer<TChangeRecord>>(std::exchange(sender.Prepared, {}))));
+    }
+
+    void ReEnqueueRecords(const TSender& sender) {
+        for (const auto& record : sender.Pending) {
+            Enqueued.insert(ReEnqueue(record));
+        }
+
+        for (const auto& record : sender.Prepared) {
+            if (!record->IsBroadcast()) {
+                Enqueued.insert(ReEnqueue(record->GetOrder(), record->GetBody().size()));
+                MemUsage -= record->GetBody().size();
+            }
+        }
+    }
+
+    TBroadcast& EnsureBroadcast(IChangeRecord::TPtr record) {
+        Y_ABORT_UNLESS(record->IsBroadcast());
+
+        auto it = Broadcasting.find(record->GetOrder());
+        if (it != Broadcasting.end()) {
+            return it->second;
+        }
+
+        THashSet<ui64> partitionIds;
+        for (const auto& [partitionId, _] : Senders) {
+            partitionIds.insert(partitionId);
+        }
+        for (const auto partitionId : GonePartitions) {
+            partitionIds.insert(partitionId);
+        }
+
+        auto res = Broadcasting.emplace(record->GetOrder(), TBroadcast{
+            .Record = {record->GetOrder(), record->GetBody().size()},
+            .Partitions = partitionIds,
+            .PendingPartitions = partitionIds,
+        });
+
+        return res.first->second;
+    }
+
+    bool AddBroadcastPartition(ui64 order, ui64 partitionId) {
+        auto it = Broadcasting.find(order);
+        Y_ABORT_UNLESS(it != Broadcasting.end());
+
+        auto& broadcast = it->second;
+        if (broadcast.Partitions.contains(partitionId)) {
+            return false;
+        }
+
+        broadcast.Partitions.insert(partitionId);
+        broadcast.PendingPartitions.insert(partitionId);
+
+        return true;
+    }
+
+    bool RemoveBroadcastPartition(ui64 order, ui64 partitionId) {
+        auto it = Broadcasting.find(order);
+        Y_ABORT_UNLESS(it != Broadcasting.end());
 
-    bool RequestRecords();
-    void SendRecords();
+        auto& broadcast = it->second;
+        broadcast.Partitions.erase(partitionId);
+        broadcast.PendingPartitions.erase(partitionId);
+        broadcast.CompletedPartitions.erase(partitionId);
 
-    void SendPreparedRecords(ui64 partitionId);
-    void ReEnqueueRecords(const TSender& sender);
+        return MaybeCompleteBroadcast(order);
+    }
+
+    bool CompleteBroadcastPartition(ui64 order, ui64 partitionId)  {
+        auto it = Broadcasting.find(order);
+        Y_ABORT_UNLESS(it != Broadcasting.end());
+
+        auto& broadcast = it->second;
+        broadcast.CompletedPartitions.insert(partitionId);
+
+        return MaybeCompleteBroadcast(order);
+    }
+
+    bool MaybeCompleteBroadcast(ui64 order) {
+        auto it = Broadcasting.find(order);
+        Y_ABORT_UNLESS(it != Broadcasting.end());
+
+        auto& broadcast = it->second;
+        if (broadcast.PendingPartitions || broadcast.Partitions.size() != broadcast.CompletedPartitions.size()) {
+            return false;
+        }
+
+        Broadcasting.erase(it);
+        return true;
+    }
 
-    TBroadcast& EnsureBroadcast(IChangeRecord::TPtr record);
-    bool AddBroadcastPartition(ui64 order, ui64 partitionId);
-    bool RemoveBroadcastPartition(ui64 order, ui64 partitionId);
-    bool CompleteBroadcastPartition(ui64 order, ui64 partitionId);
-    bool MaybeCompleteBroadcast(ui64 order);
     void ProcessBroadcasting(std::function<bool(TBaseChangeSender*, ui64, ui64)> f,
-        ui64 partitionId, const TVector<ui64>& broadcasting);
+        ui64 partitionId, const TVector<ui64>& broadcasting)
+    {
+        TVector<ui64> remove;
+        for (const auto order : broadcasting) {
+            if (std::invoke(f, this, order, partitionId)) {
+                remove.push_back(order);
+            }
+        }
+
+        if (remove) {
+            RemoveRecords(std::move(remove));
+        }
+    }
 
 protected:
     template <typename T>
@@ -154,25 +412,358 @@ protected:
         ActorOps->Send(GetChangeServer(), new TEvChangeExchange::TEvRemoveRecords(std::move(records)));
     }
 
-    void CreateSenders(const TVector<ui64>& partitionIds, bool partitioningChanged = true) override;
-    void KillSenders() override;
-    void RemoveRecords() override;
+    TActorId GetChangeServer() const { return ChangeServer; }
+    void CreateSenders(const TVector<ui64>& partitionIds, bool partitioningChanged = true) {
+        if (partitioningChanged) {
+            CreateMissingSenders(partitionIds);
+        } else {
+            RecreateSenders(GonePartitions);
+        }
+
+        GonePartitions.clear();
+
+        if (!Enqueued || !RequestRecords()) {
+            SendRecords();
+        }
+    }
+
+    void KillSenders() {
+        for (const auto& [_, sender] : std::exchange(Senders, {})) {
+            if (sender.ActorId) {
+                ActorOps->Send(sender.ActorId, new TEvents::TEvPoisonPill());
+            }
+        }
+    }
+
+    void RemoveRecords() {
+        THashSet<ui64> remove;
+
+        for (const auto& record : std::exchange(Enqueued, {})) {
+            remove.insert(record.Order);
+        }
+
+        for (const auto& record : std::exchange(PendingBody, {})) {
+            remove.insert(record.Order);
+        }
+
+        for (const auto& [order, _] : std::exchange(PendingSent, {})) {
+            remove.insert(order);
+        }
+
+        for (const auto& [order, _] : std::exchange(Broadcasting, {})) {
+            remove.insert(order);
+        }
+
+        for (const auto& [_, sender] : Senders) {
+            for (const auto& record : sender.Pending) {
+                remove.insert(record.Order);
+            }
+
+            for (const auto& record : sender.Prepared) {
+                remove.insert(record->GetOrder());
+            }
+        }
+
+        if (remove) {
+            RemoveRecords(TVector<ui64>(remove.begin(), remove.end()));
+        }
+    }
+
+    void EnqueueRecords(TVector<TEvChangeExchange::TEvEnqueueRecords::TRecordInfo>&& records) {
+        for (auto& record : records) {
+            Y_VERIFY_S(PathId == record.PathId, "Unexpected record's path id"
+                << ": expected# " << PathId
+                << ", got# " << record.PathId);
+            Enqueued.emplace(record.Order, record.BodySize);
+        }
+
+        RequestRecords();
+    }
+
+    void ProcessRecords(TVector<typename TChangeRecord::TPtr>&& records) {
+        for (auto& record : records) {
+            auto it = PendingBody.find(record->GetOrder());
+            if (it == PendingBody.end()) {
+                continue;
+            }
+
+            if (it->BodySize != record->GetBody().size()) {
+                MemUsage -= it->BodySize;
+                MemUsage += record->GetBody().size();
+            }
+
+            if (record->IsBroadcast()) {
+                // assume that broadcast records are too small to affect memory consumption
+                MemUsage -= record->GetBody().size();
+            }
+
+            PendingSent.emplace(record->GetOrder(), std::move(record));
+            PendingBody.erase(it);
+        }
+
+        SendRecords();
+    }
+
+    void ForgetRecords(TVector<ui64>&& records) {
+        for (const auto& record : records) {
+            auto it = PendingBody.find(record);
+            if (it == PendingBody.end()) {
+                continue;
+            }
+
+            MemUsage -= it->BodySize;
+            PendingBody.erase(it);
+        }
+
+        RequestRecords();
+    }
+
+    void OnReady(ui64 partitionId) {
+        auto it = Senders.find(partitionId);
+        if (it == Senders.end()) {
+            return;
+        }
+
+        auto& sender = it->second;
+        sender.Ready = true;
 
-    void EnqueueRecords(TVector<TEvChangeExchange::TEvEnqueueRecords::TRecordInfo>&& records) override;
-    void ProcessRecords(TVector<IChangeRecord::TPtr>&& records) override;
-    void ForgetRecords(TVector<ui64>&& records) override;
-    void OnReady(ui64 partitionId) override;
-    void OnGone(ui64 partitionId) override;
+        if (sender.Pending) {
+            RemoveRecords(std::exchange(sender.Pending, {}));
+        }
+
+        if (sender.Broadcasting) {
+            ProcessBroadcasting(&TBaseChangeSender::CompleteBroadcastPartition,
+                partitionId, std::exchange(sender.Broadcasting, {}));
+        }
 
-    explicit TBaseChangeSender(IActorOps* actorOps, IChangeSenderResolver* resolver, const TPathId& pathId);
+        if (sender.Prepared) {
+            SendPreparedRecords(partitionId);
+        }
+
+        RequestRecords();
+    }
+
+    void OnGone(ui64 partitionId) {
+        auto it = Senders.find(partitionId);
+        if (it == Senders.end()) {
+            return;
+        }
 
-    void RenderHtmlPage(ui64 tabletId, NMon::TEvRemoteHttpInfo::TPtr& ev, const TActorContext& ctx);
+        ReEnqueueRecords(it->second);
+        Senders.erase(it);
+        GonePartitions.push_back(partitionId);
+
+        if (Resolver->IsResolving()) {
+            return;
+        }
+
+        Resolver->Resolve();
+    }
+
+    explicit TBaseChangeSender(
+        IActorOps* const actorOps,
+        IChangeSenderResolver* const resolver,
+        ISenderFactory* const senderFactory,
+        const TActorId changeServer,
+        const TPathId& pathId)
+            : ActorOps(actorOps)
+            , Resolver(resolver)
+            , SenderFactory(senderFactory)
+            , ChangeServer(changeServer)
+            , PathId(pathId)
+            , MemLimit(192_KB)
+            , MemUsage(0)
+    {}
+
+    void RenderHtmlPage(ui64 tabletId, NMon::TEvRemoteHttpInfo::TPtr& ev, const TActorContext& ctx) {
+        const auto& cgi = ev->Get()->Cgi();
+        if (const auto& str = cgi.Get("partitionId")) {
+            ui64 partitionId = 0;
+            if (TryFromString(str, partitionId)) {
+                auto it = Senders.find(partitionId);
+                if (it != Senders.end()) {
+                    if (const auto& to = it->second.ActorId) {
+                        ctx.Send(ev->Forward(to));
+                    } else {
+                        ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes(TStringBuilder()
+                            << "Change sender '" << PathId << ":" << partitionId << "' is not running"));
+                    }
+                } else {
+                    ActorOps->Send(ev->Sender, new NMon::TEvRemoteBinaryInfoRes(NMonitoring::HTTPNOTFOUND));
+                }
+            } else {
+                ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes("Invalid partitionId"));
+            }
+
+            return;
+        }
+
+        TStringStream html;
+
+        HTML(html) {
+            Header(html, "Change sender", tabletId);
+
+            SimplePanel(html, "Info", [this](IOutputStream& html) {
+                HTML(html) {
+                    DL_CLASS("dl-horizontal") {
+                        TermDesc(html, "MemLimit", MemLimit);
+                        TermDesc(html, "MemUsage", MemUsage);
+                    }
+                }
+            });
+
+            SimplePanel(html, "Partition senders", [this, tabletId](IOutputStream& html) {
+                HTML(html) {
+                    TABLE_CLASS("table table-hover") {
+                        TABLEHEAD() {
+                            TABLER() {
+                                TABLEH() { html << "#"; }
+                                TABLEH() { html << "PartitionId"; }
+                                TABLEH() { html << "Ready"; }
+                                TABLEH() { html << "Pending"; }
+                                TABLEH() { html << "Prepared"; }
+                                TABLEH() { html << "Broadcasting"; }
+                                TABLEH() { html << "Actor"; }
+                            }
+                        }
+                        TABLEBODY() {
+                            ui32 i = 0;
+                            for (const auto& [partitionId, sender] : Senders) {
+                                TABLER() {
+                                    TABLED() { html << ++i; }
+                                    TABLED() { html << partitionId; }
+                                    TABLED() { html << sender.Ready; }
+                                    TABLED() { html << sender.Pending.size(); }
+                                    TABLED() { html << sender.Prepared.size(); }
+                                    TABLED() { html << sender.Broadcasting.size(); }
+                                    TABLED() { ActorLink(html, tabletId, PathId, partitionId); }
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+
+            CollapsedPanel(html, "Enqueued", "enqueued", [this](IOutputStream& html) {
+                HTML(html) {
+                    TABLE_CLASS("table table-hover") {
+                        TABLEHEAD() {
+                            TABLER() {
+                                TABLEH() { html << "#"; }
+                                TABLEH() { html << "Order"; }
+                                TABLEH() { html << "BodySize"; }
+                            }
+                        }
+                        TABLEBODY() {
+                            ui32 i = 0;
+                            for (const auto& record : Enqueued) {
+                                TABLER() {
+                                    TABLED() { html << ++i; }
+                                    TABLED() { html << record.Order; }
+                                    TABLED() { html << record.BodySize; }
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+
+            CollapsedPanel(html, "PendingBody", "pendingBody", [this](IOutputStream& html) {
+                HTML(html) {
+                    TABLE_CLASS("table table-hover") {
+                        TABLEHEAD() {
+                            TABLER() {
+                                TABLEH() { html << "#"; }
+                                TABLEH() { html << "Order"; }
+                                TABLEH() { html << "BodySize"; }
+                            }
+                        }
+                        TABLEBODY() {
+                            ui32 i = 0;
+                            for (const auto& record : PendingBody) {
+                                TABLER() {
+                                    TABLED() { html << ++i; }
+                                    TABLED() { html << record.Order; }
+                                    TABLED() { html << record.BodySize; }
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+
+            CollapsedPanel(html, "PendingSent", "pendingSent", [this](IOutputStream& html) {
+                HTML(html) {
+                    TABLE_CLASS("table table-hover") {
+                        TABLEHEAD() {
+                            TABLER() {
+                                TABLEH() { html << "#"; }
+                                TABLEH() { html << "Order"; }
+                                TABLEH() { html << "Group"; }
+                                TABLEH() { html << "Step"; }
+                                TABLEH() { html << "TxId"; }
+                                TABLEH() { html << "Kind"; }
+                                TABLEH() { html << "Source"; }
+                            }
+                        }
+                        TABLEBODY() {
+                            ui32 i = 0;
+                            for (const auto& [order, record] : PendingSent) {
+                                TABLER() {
+                                    TABLED() { html << ++i; }
+                                    TABLED() { html << order; }
+                                    TABLED() { html << record->GetGroup(); }
+                                    TABLED() { html << record->GetStep(); }
+                                    TABLED() { html << record->GetTxId(); }
+                                    TABLED() { html << record->GetKind(); }
+                                    TABLED() { html << record->GetSource(); }
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+
+            CollapsedPanel(html, "Broadcasting", "broadcasting", [this](IOutputStream& html) {
+                HTML(html) {
+                    TABLE_CLASS("table table-hover") {
+                        TABLEHEAD() {
+                            TABLER() {
+                                TABLEH() { html << "#"; }
+                                TABLEH() { html << "Order"; }
+                                TABLEH() { html << "BodySize"; }
+                                TABLEH() { html << "Partitions"; }
+                                TABLEH() { html << "PendingPartitions"; }
+                                TABLEH() { html << "CompletedPartitions"; }
+                            }
+                        }
+                        TABLEBODY() {
+                            ui32 i = 0;
+                            for (const auto& [order, broadcast] : Broadcasting) {
+                                TABLER() {
+                                    TABLED() { html << ++i; }
+                                    TABLED() { html << order; }
+                                    TABLED() { html << broadcast.Record.BodySize; }
+                                    TABLED() { html << broadcast.Partitions.size(); }
+                                    TABLED() { html << broadcast.PendingPartitions.size(); }
+                                    TABLED() { html << broadcast.CompletedPartitions.size(); }
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+        }
+
+        ActorOps->Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes(html.Str()));
+    }
 
 private:
     IActorOps* const ActorOps;
     IChangeSenderResolver* const Resolver;
-
+    ISenderFactory* const SenderFactory;
 protected:
+    TActorId ChangeServer;
     const TPathId PathId;
 
 private:
@@ -182,11 +773,10 @@ private:
     THashMap<ui64, TSender> Senders; // ui64 is partition id
     TSet<TEnqueuedRecord> Enqueued;
     TSet<TIncompleteRecord> PendingBody;
-    TMap<ui64, IChangeRecord::TPtr> PendingSent; // ui64 is order
+    TMap<ui64, typename TChangeRecord::TPtr> PendingSent; // ui64 is order
     THashMap<ui64, TBroadcast> Broadcasting; // ui64 is order
 
     TVector<ui64> GonePartitions;
-
 }; // TBaseChangeSender
 
 }
diff --git a/ydb/core/change_exchange/change_sender_resolver.h b/ydb/core/change_exchange/change_sender_resolver.h
new file mode 100644
index 00000000000..71cb67605a9
--- /dev/null
+++ b/ydb/core/change_exchange/change_sender_resolver.h
@@ -0,0 +1,24 @@
+#pragma once
+
+#include <ydb/core/protos/flat_scheme_op.pb.h>
+#include <ydb/core/scheme/scheme_tabledefs.h>
+#include <ydb/core/scheme_types/scheme_type_info.h>
+
+#include <util/generic/vector.h>
+
+namespace NKikimr::NChangeExchange {
+
+class IChangeSenderResolver {
+public:
+    virtual ~IChangeSenderResolver() = default;
+
+    virtual void Resolve() = 0;
+    virtual bool IsResolving() const = 0;
+    virtual bool IsResolved() const = 0;
+
+    virtual const TVector<TKeyDesc::TPartitionInfo>& GetPartitions() const = 0;
+    virtual const TVector<NScheme::TTypeInfo>& GetSchema() const = 0;
+    virtual NKikimrSchemeOp::ECdcStreamFormat GetStreamFormat() const = 0;
+};
+
+} // NKikimr::NChangeExchange
diff --git a/ydb/core/change_exchange/ya.make b/ydb/core/change_exchange/ya.make
index 6e82179d832..b95ab217844 100644
--- a/ydb/core/change_exchange/ya.make
+++ b/ydb/core/change_exchange/ya.make
@@ -3,7 +3,6 @@ LIBRARY()
 SRCS(
     change_exchange.cpp
     change_record.cpp
-    change_sender_common_ops.cpp
     change_sender_monitoring.cpp
 )
 
diff --git a/ydb/core/scheme/scheme_tabledefs.h b/ydb/core/scheme/scheme_tabledefs.h
index bbe8f1cd99e..2e8dc6941b4 100644
--- a/ydb/core/scheme/scheme_tabledefs.h
+++ b/ydb/core/scheme/scheme_tabledefs.h
@@ -712,6 +712,18 @@ public:
         , Status(EStatus::Unknown)
         , Partitioning(std::make_shared<TVector<TKeyDesc::TPartitionInfo>>())
     {}
+
+    static THolder<TKeyDesc> CreateMiniKeyDesc(const TVector<NScheme::TTypeInfo> &keyColumnTypes) {
+        return THolder<TKeyDesc>(new TKeyDesc(keyColumnTypes));
+    }
+private:
+    TKeyDesc(const TVector<NScheme::TTypeInfo> &keyColumnTypes)
+        : RowOperation(ERowOperation::Unknown)
+        , KeyColumnTypes(keyColumnTypes.begin(), keyColumnTypes.end())
+        , Reverse(false)
+        , Status(EStatus::Unknown)
+        , Partitioning(std::make_shared<TVector<TKeyDesc::TPartitionInfo>>())
+    {}
 };
 
 struct TSystemColumnInfo {
diff --git a/ydb/core/tx/datashard/cdc_stream_heartbeat.cpp b/ydb/core/tx/datashard/cdc_stream_heartbeat.cpp
index 0560136f72a..6473bc62ba6 100644
--- a/ydb/core/tx/datashard/cdc_stream_heartbeat.cpp
+++ b/ydb/core/tx/datashard/cdc_stream_heartbeat.cpp
@@ -51,7 +51,7 @@ public:
                 .WithSchemaVersion(0) // not used
                 .Build();
 
-            const auto& record = *recordPtr->Get<TChangeRecord>();
+            const auto& record = *recordPtr;
             Self->PersistChangeRecord(db, record);
 
             ChangeRecords.push_back(IDataShardChangeCollector::TChange{
diff --git a/ydb/core/tx/datashard/cdc_stream_scan.cpp b/ydb/core/tx/datashard/cdc_stream_scan.cpp
index 4bae27518d4..944f9de86c2 100644
--- a/ydb/core/tx/datashard/cdc_stream_scan.cpp
+++ b/ydb/core/tx/datashard/cdc_stream_scan.cpp
@@ -314,7 +314,7 @@ public:
                 .WithSource(TChangeRecord::ESource::InitialScan)
                 .Build();
 
-            const auto& record = *recordPtr->Get<TChangeRecord>();
+            const auto& record = *recordPtr;
             Self->PersistChangeRecord(db, record);
 
             ChangeRecords.push_back(IDataShardChangeCollector::TChange{
diff --git a/ydb/core/tx/datashard/change_collector.cpp b/ydb/core/tx/datashard/change_collector.cpp
index c2f30163d44..56f4e673ba5 100644
--- a/ydb/core/tx/datashard/change_collector.cpp
+++ b/ydb/core/tx/datashard/change_collector.cpp
@@ -141,7 +141,7 @@ public:
             .WithBody(body.SerializeAsString())
             .Build();
 
-        const auto& record = *recordPtr->Get<TChangeRecord>();
+        const auto& record = *recordPtr;
         Self->PersistChangeRecord(db, record);
 
         if (record.GetLockId() == 0) {
diff --git a/ydb/core/tx/datashard/change_record.h b/ydb/core/tx/datashard/change_record.h
index 236f7d4ce46..0366fd2c77d 100644
--- a/ydb/core/tx/datashard/change_record.h
+++ b/ydb/core/tx/datashard/change_record.h
@@ -3,10 +3,14 @@
 #include "datashard_user_table.h"
 
 #include <ydb/core/change_exchange/change_record.h>
+#include <ydb/core/change_exchange/change_sender_resolver.h>
+#include <ydb/core/change_exchange/change_exchange.h>
 #include <ydb/core/scheme/scheme_pathid.h>
 #include <ydb/core/scheme/scheme_tablecell.h>
+#include <ydb/services/lib/sharding/sharding.h>
 
 #include <util/generic/maybe.h>
+#include <util/string/join.h>
 
 namespace NKikimrChangeExchange {
     class TChangeRecord;
@@ -20,6 +24,8 @@ class TChangeRecord: public NChangeExchange::TChangeRecordBase {
     friend class TChangeRecordBuilder;
 
 public:
+    using TPtr = TIntrusivePtr<TChangeRecord>;
+
     ui64 GetGroup() const override { return Group; }
     ui64 GetStep() const override { return Step; }
     ui64 GetTxId() const override { return TxId; }
@@ -42,6 +48,50 @@ public:
 
     void Out(IOutputStream& out) const override;
 
+    ui64 ResolvePartitionId(NChangeExchange::IChangeSenderResolver* const resolver) const override {
+        const auto& partitions = resolver->GetPartitions();
+        Y_ABORT_UNLESS(partitions);
+        const auto& schema = resolver->GetSchema();
+        const auto streamFormat = resolver->GetStreamFormat();
+
+        switch (streamFormat) {
+            case NKikimrSchemeOp::ECdcStreamFormatProto: {
+                const auto range = TTableRange(GetKey());
+                Y_ABORT_UNLESS(range.Point);
+
+                const auto it = LowerBound(
+                    partitions.cbegin(), partitions.cend(), true,
+                    [&](const auto& partition, bool) {
+                        Y_ABORT_UNLESS(partition.Range);
+                        const int compares = CompareBorders<true, false>(
+                            partition.Range->EndKeyPrefix.GetCells(), range.From,
+                            partition.Range->IsInclusive || partition.Range->IsPoint,
+                            range.InclusiveFrom || range.Point, schema
+                        );
+
+                        return (compares < 0);
+                    }
+                );
+
+                Y_ABORT_UNLESS(it != partitions.cend());
+                return it->ShardId;
+            }
+
+            case NKikimrSchemeOp::ECdcStreamFormatJson:
+            case NKikimrSchemeOp::ECdcStreamFormatDynamoDBStreamsJson:
+            case NKikimrSchemeOp::ECdcStreamFormatDebeziumJson: {
+                using namespace NKikimr::NDataStreams::V1;
+                const auto hashKey = HexBytesToDecimal(GetPartitionKey() /* MD5 */);
+                return ShardFromDecimal(hashKey, partitions.size());
+            }
+
+            default: {
+                Y_FAIL_S("Unknown format"
+                    << ": format# " << static_cast<int>(streamFormat));
+            }
+        }
+    }
+
 private:
     ui64 Group = 0;
     ui64 Step = 0;
@@ -119,6 +169,31 @@ public:
 
 }
 
+namespace NKikimr {
+
+template <>
+struct TChangeRecordContainer<NDataShard::TChangeRecord>
+    : public TBaseChangeRecordContainer
+{
+    TChangeRecordContainer() = default;
+
+    explicit TChangeRecordContainer(TVector<NDataShard::TChangeRecord::TPtr>&& records)
+        : Records(std::move(records))
+    {}
+
+    TVector<NDataShard::TChangeRecord::TPtr> Records;
+
+    TString Out() override {
+        return TStringBuilder() << "[" << JoinSeq(",", Records) << "]";
+    }
+};
+
+}
+
 Y_DECLARE_OUT_SPEC(inline, NKikimr::NDataShard::TChangeRecord, out, value) {
     return value.Out(out);
 }
+
+Y_DECLARE_OUT_SPEC(inline, NKikimr::NDataShard::TChangeRecord::TPtr, out, value) {
+    return value->Out(out);
+}
diff --git a/ydb/core/tx/datashard/change_sender_async_index.cpp b/ydb/core/tx/datashard/change_sender_async_index.cpp
index c0630662a4e..38492b20728 100644
--- a/ydb/core/tx/datashard/change_sender_async_index.cpp
+++ b/ydb/core/tx/datashard/change_sender_async_index.cpp
@@ -128,8 +128,10 @@ class TAsyncIndexChangeSenderShard: public TActorBootstrapped<TAsyncIndexChangeS
         records->Record.SetOrigin(DataShard.TabletId);
         records->Record.SetGeneration(DataShard.Generation);
 
-        for (auto recordPtr : ev->Get()->Records) {
-            const auto& record = *recordPtr->Get<TChangeRecord>();
+        auto& evRecords = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
+
+        for (auto& recordPtr : evRecords) {
+            const auto& record = *recordPtr;
 
             if (record.GetOrder() <= LastRecordOrder) {
                 continue;
@@ -330,8 +332,9 @@ private:
 
 class TAsyncIndexChangeSenderMain
     : public TActorBootstrapped<TAsyncIndexChangeSenderMain>
-    , public NChangeExchange::TBaseChangeSender
+    , public NChangeExchange::TBaseChangeSender<TChangeRecord>
     , public NChangeExchange::IChangeSenderResolver
+    , public NChangeExchange::ISenderFactory
     , private NSchemeCache::TSchemeCacheHelpers
 {
     TStringBuf GetLogPrefix() const {
@@ -704,10 +707,6 @@ class TAsyncIndexChangeSenderMain
         return StateBase(ev);
     }
 
-    TActorId GetChangeServer() const override {
-        return DataShard.ActorId;
-    }
-
     void Resolve() override {
         ResolveIndex();
     }
@@ -716,31 +715,11 @@ class TAsyncIndexChangeSenderMain
         return KeyDesc && KeyDesc->GetPartitions();
     }
 
-    ui64 GetPartitionId(NChangeExchange::IChangeRecord::TPtr record) const override {
-        Y_ABORT_UNLESS(KeyDesc);
-        Y_ABORT_UNLESS(KeyDesc->GetPartitions());
-
-        const auto range = TTableRange(record->Get<TChangeRecord>()->GetKey());
-        Y_ABORT_UNLESS(range.Point);
-
-        TVector<TKeyDesc::TPartitionInfo>::const_iterator it = LowerBound(
-            KeyDesc->GetPartitions().begin(), KeyDesc->GetPartitions().end(), true,
-            [&](const TKeyDesc::TPartitionInfo& partition, bool) {
-                const int compares = CompareBorders<true, false>(
-                    partition.Range->EndKeyPrefix.GetCells(), range.From,
-                    partition.Range->IsInclusive || partition.Range->IsPoint,
-                    range.InclusiveFrom || range.Point, KeyDesc->KeyColumnTypes
-                );
-
-                return (compares < 0);
-            }
-        );
-
-        Y_ABORT_UNLESS(it != KeyDesc->GetPartitions().end());
-        return it->ShardId; // partition = shard
-    }
+    const TVector<TKeyDesc::TPartitionInfo>& GetPartitions() const override { return KeyDesc->GetPartitions(); }
+    const TVector<NScheme::TTypeInfo>& GetSchema() const override { return KeyDesc->KeyColumnTypes; }
+    NKikimrSchemeOp::ECdcStreamFormat GetStreamFormat() const override { return NKikimrSchemeOp::ECdcStreamFormatProto; }
 
-    IActor* CreateSender(ui64 partitionId) override {
+    IActor* CreateSender(ui64 partitionId) const override {
         return new TAsyncIndexChangeSenderShard(SelfId(), DataShard, partitionId, IndexTablePathId, TagMap);
     }
 
@@ -751,7 +730,8 @@ class TAsyncIndexChangeSenderMain
 
     void Handle(NChangeExchange::TEvChangeExchange::TEvRecords::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
-        ProcessRecords(std::move(ev->Get()->Records));
+        auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
+        ProcessRecords(std::move(records));
     }
 
     void Handle(NChangeExchange::TEvChangeExchange::TEvForgetRecords::TPtr& ev) {
@@ -798,7 +778,7 @@ public:
 
     explicit TAsyncIndexChangeSenderMain(const TDataShardId& dataShard, const TTableId& userTableId, const TPathId& indexPathId)
         : TActorBootstrapped()
-        , TBaseChangeSender(this, this, indexPathId)
+        , TBaseChangeSender(this, this, this, dataShard.ActorId, indexPathId)
         , DataShard(dataShard)
         , UserTableId(userTableId)
         , IndexTableVersion(0)
diff --git a/ydb/core/tx/datashard/change_sender_cdc_stream.cpp b/ydb/core/tx/datashard/change_sender_cdc_stream.cpp
index 55cf00095e5..5300357c24c 100644
--- a/ydb/core/tx/datashard/change_sender_cdc_stream.cpp
+++ b/ydb/core/tx/datashard/change_sender_cdc_stream.cpp
@@ -95,8 +95,9 @@ class TCdcChangeSenderPartition: public TActorBootstrapped<TCdcChangeSenderParti
         LOG_D("Handle " << ev->Get()->ToString());
         NKikimrClient::TPersQueueRequest request;
 
-        for (auto recordPtr : ev->Get()->Records) {
-            const auto& record = *recordPtr->Get<TChangeRecord>();
+        auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
+        for (auto recordPtr : records) {
+            const auto& record = *recordPtr;
 
             if (record.GetSeqNo() <= MaxSeqNo) {
                 continue;
@@ -294,8 +295,9 @@ private:
 
 class TCdcChangeSenderMain
     : public TActorBootstrapped<TCdcChangeSenderMain>
-    , public NChangeExchange::TBaseChangeSender
+    , public NChangeExchange::TBaseChangeSender<TChangeRecord>
     , public NChangeExchange::IChangeSenderResolver
+    , public NChangeExchange::ISenderFactory
     , private NSchemeCache::TSchemeCacheHelpers
 {
     struct TPQPartitionInfo {
@@ -337,31 +339,6 @@ class TCdcChangeSenderMain
 
     }; // TPQPartitionInfo
 
-    struct TKeyDesc {
-        struct TPartitionInfo {
-            ui32 PartitionId;
-            ui64 ShardId;
-            TSerializedCellVec EndKeyPrefix;
-            // just a hint
-            static constexpr bool IsInclusive = false;
-            static constexpr bool IsPoint = false;
-
-            explicit TPartitionInfo(const TPQPartitionInfo& info)
-                : PartitionId(info.PartitionId)
-                , ShardId(info.ShardId)
-            {
-                if (info.KeyRange.ToBound) {
-                    EndKeyPrefix = *info.KeyRange.ToBound;
-                }
-            }
-
-        }; // TPartitionInfo
-
-        TVector<NScheme::TTypeInfo> Schema;
-        TVector<TPartitionInfo> Partitions;
-
-    }; // TKeyDesc
-
     TStringBuf GetLogPrefix() const {
         if (!LogPrefix) {
             LogPrefix = TStringBuilder()
@@ -453,11 +430,11 @@ class TCdcChangeSenderMain
         return false;
     }
 
-    static TVector<ui64> MakePartitionIds(const TVector<TKeyDesc::TPartitionInfo>& partitions) {
+    static TVector<ui64> MakePartitionIds(const TVector<NKikimr::TKeyDesc::TPartitionInfo>& partitions) {
         TVector<ui64> result(Reserve(partitions.size()));
 
         for (const auto& partition : partitions) {
-            result.push_back(partition.PartitionId);
+            result.push_back(partition.ShardId);
         }
 
         return result;
@@ -587,16 +564,16 @@ class TCdcChangeSenderMain
         const auto& pqDesc = entry.PQGroupInfo->Description;
         const auto& pqConfig = pqDesc.GetPQTabletConfig();
 
-        KeyDesc = MakeHolder<TKeyDesc>();
+        TVector<NScheme::TTypeInfo> schema;
         PartitionToShard.clear();
 
-        KeyDesc->Schema.reserve(pqConfig.PartitionKeySchemaSize());
+        schema.reserve(pqConfig.PartitionKeySchemaSize());
         for (const auto& keySchema : pqConfig.GetPartitionKeySchema()) {
             // TODO: support pg types
-            KeyDesc->Schema.push_back(NScheme::TTypeInfo(keySchema.GetTypeId()));
+            schema.push_back(NScheme::TTypeInfo(keySchema.GetTypeId()));
         }
 
-        TSet<TPQPartitionInfo, TPQPartitionInfo::TLess> partitions(KeyDesc->Schema);
+        TSet<TPQPartitionInfo, TPQPartitionInfo::TLess> partitions(schema);
         THashSet<ui64> shards;
 
         for (const auto& partition : pqDesc.GetPartitions()) {
@@ -606,8 +583,8 @@ class TCdcChangeSenderMain
             PartitionToShard.emplace(partitionId, shardId);
 
             auto keyRange = TPartitionKeyRange::Parse(partition.GetKeyRange());
-            Y_ABORT_UNLESS(!keyRange.FromBound || keyRange.FromBound->GetCells().size() == KeyDesc->Schema.size());
-            Y_ABORT_UNLESS(!keyRange.ToBound || keyRange.ToBound->GetCells().size() == KeyDesc->Schema.size());
+            Y_ABORT_UNLESS(!keyRange.FromBound || keyRange.FromBound->GetCells().size() == schema.size());
+            Y_ABORT_UNLESS(!keyRange.ToBound || keyRange.ToBound->GetCells().size() == schema.size());
 
             partitions.insert({partitionId, shardId, std::move(keyRange)});
             shards.insert(shardId);
@@ -617,7 +594,8 @@ class TCdcChangeSenderMain
         bool isFirst = true;
         const TPQPartitionInfo* prev = nullptr;
 
-        KeyDesc->Partitions.reserve(partitions.size());
+        TVector<NKikimr::TKeyDesc::TPartitionInfo> partitioning;
+        partitioning.reserve(partitions.size());
         for (const auto& cur : partitions) {
             if (isFirst) {
                 isFirst = false;
@@ -629,7 +607,16 @@ class TCdcChangeSenderMain
                 // TODO: compare cells
             }
 
-            KeyDesc->Partitions.emplace_back(cur);
+            auto& part = partitioning.emplace_back(cur.PartitionId); // TODO: double-check that it is right partitioning
+
+            if (cur.KeyRange.ToBound) {
+                part.Range = NKikimr::TKeyDesc::TPartitionRangeInfo{
+                    .EndKeyPrefix = *cur.KeyRange.ToBound,
+                };
+            } else {
+                part.Range = NKikimr::TKeyDesc::TPartitionRangeInfo{};
+            }
+
             prev = &cur;
         }
 
@@ -641,7 +628,10 @@ class TCdcChangeSenderMain
         const bool versionChanged = !TopicVersion || TopicVersion != topicVersion;
         TopicVersion = topicVersion;
 
-        CreateSenders(MakePartitionIds(KeyDesc->Partitions), versionChanged);
+        KeyDesc = NKikimr::TKeyDesc::CreateMiniKeyDesc(schema);
+        KeyDesc->Partitioning = std::make_shared<TVector<NKikimr::TKeyDesc::TPartitionInfo>>(std::move(partitioning));
+
+        CreateSenders(MakePartitionIds(*KeyDesc->Partitioning), versionChanged);
         Become(&TThis::StateMain);
     }
 
@@ -651,60 +641,19 @@ class TCdcChangeSenderMain
         return StateBase(ev);
     }
 
-    TActorId GetChangeServer() const override {
-        return DataShard.ActorId;
-    }
-
     void Resolve() override {
         ResolveCdcStream();
     }
 
     bool IsResolved() const override {
-        return KeyDesc && KeyDesc->Partitions;
+        return KeyDesc && KeyDesc->Partitioning;
     }
 
-    ui64 GetPartitionId(NChangeExchange::IChangeRecord::TPtr record) const override {
-        Y_ABORT_UNLESS(KeyDesc);
-        Y_ABORT_UNLESS(KeyDesc->Partitions);
-
-        switch (Stream.Format) {
-            case NKikimrSchemeOp::ECdcStreamFormatProto: {
-                const auto range = TTableRange(record->Get<TChangeRecord>()->GetKey());
-                Y_ABORT_UNLESS(range.Point);
-
-                TVector<TKeyDesc::TPartitionInfo>::const_iterator it = LowerBound(
-                    KeyDesc->Partitions.begin(), KeyDesc->Partitions.end(), true,
-                    [&](const TKeyDesc::TPartitionInfo& partition, bool) {
-                        const int compares = CompareBorders<true, false>(
-                            partition.EndKeyPrefix.GetCells(), range.From,
-                            partition.IsInclusive || partition.IsPoint,
-                            range.InclusiveFrom || range.Point, KeyDesc->Schema
-                        );
-
-                        return (compares < 0);
-                    }
-                );
-
-                Y_ABORT_UNLESS(it != KeyDesc->Partitions.end());
-                return it->PartitionId;
-            }
-
-            case NKikimrSchemeOp::ECdcStreamFormatJson:
-            case NKikimrSchemeOp::ECdcStreamFormatDynamoDBStreamsJson:
-            case NKikimrSchemeOp::ECdcStreamFormatDebeziumJson: {
-                using namespace NKikimr::NDataStreams::V1;
-                const auto hashKey = HexBytesToDecimal(record->Get<TChangeRecord>()->GetPartitionKey() /* MD5 */);
-                return ShardFromDecimal(hashKey, KeyDesc->Partitions.size());
-            }
-
-            default: {
-                Y_FAIL_S("Unknown format"
-                    << ": format# " << static_cast<int>(Stream.Format));
-            }
-        }
-    }
+    const TVector<NKikimr::TKeyDesc::TPartitionInfo>& GetPartitions() const override { return KeyDesc->GetPartitions(); }
+    const TVector<NScheme::TTypeInfo>& GetSchema() const override { return KeyDesc->KeyColumnTypes; }
+    NKikimrSchemeOp::ECdcStreamFormat GetStreamFormat() const override { return Stream.Format; }
 
-    IActor* CreateSender(ui64 partitionId) override {
+    IActor* CreateSender(ui64 partitionId) const override {
         Y_ABORT_UNLESS(PartitionToShard.contains(partitionId));
         const auto shardId = PartitionToShard.at(partitionId);
         return new TCdcChangeSenderPartition(SelfId(), DataShard, partitionId, shardId, Stream);
@@ -717,7 +666,8 @@ class TCdcChangeSenderMain
 
     void Handle(NChangeExchange::TEvChangeExchange::TEvRecords::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
-        ProcessRecords(std::move(ev->Get()->Records));
+        auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
+        ProcessRecords(std::move(records));
     }
 
     void Handle(NChangeExchange::TEvChangeExchange::TEvForgetRecords::TPtr& ev) {
@@ -764,7 +714,7 @@ public:
 
     explicit TCdcChangeSenderMain(const TDataShardId& dataShard, const TPathId& streamPathId)
         : TActorBootstrapped()
-        , TBaseChangeSender(this, this, streamPathId)
+        , TBaseChangeSender(this, this, this, dataShard.ActorId, streamPathId)
         , DataShard(dataShard)
         , TopicVersion(0)
     {
@@ -803,7 +753,7 @@ private:
     TUserTable::TCdcStream Stream;
     TPathId TopicPathId;
     ui64 TopicVersion;
-    THolder<TKeyDesc> KeyDesc;
+    THolder<NKikimr::TKeyDesc> KeyDesc;
     THashMap<ui32, ui64> PartitionToShard;
 
 }; // TCdcChangeSenderMain
diff --git a/ydb/core/tx/datashard/datashard_change_sending.cpp b/ydb/core/tx/datashard/datashard_change_sending.cpp
index eb6486cae78..181f3fdf8d2 100644
--- a/ydb/core/tx/datashard/datashard_change_sending.cpp
+++ b/ydb/core/tx/datashard/datashard_change_sending.cpp
@@ -51,7 +51,7 @@ class TDataShard::TTxRequestChangeRecords: public TTransactionBase<TDataShard> {
 
     struct TLoadResult {
         NTable::EReady Ready;
-        IChangeRecord::TPtr Record;
+        TIntrusivePtr<NKikimr::NDataShard::TChangeRecord> Record;
 
         TLoadResult() = default;
         TLoadResult(NTable::EReady ready)
@@ -59,7 +59,7 @@ class TDataShard::TTxRequestChangeRecords: public TTransactionBase<TDataShard> {
         {
         }
 
-        explicit TLoadResult(NTable::EReady ready, IChangeRecord::TPtr record)
+        explicit TLoadResult(NTable::EReady ready, TIntrusivePtr<NKikimr::NDataShard::TChangeRecord> record)
             : Ready(ready)
             , Record(record)
         {
@@ -233,7 +233,7 @@ public:
             LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, "Send " << records.size() << " change records"
                 << ": to# " << to
                 << ", at tablet# " << Self->TabletID());
-            ctx.Send(to, new NChangeExchange::TEvChangeExchange::TEvRecords(std::move(records)));
+            ctx.Send(to, new NChangeExchange::TEvChangeExchange::TEvRecords(std::make_shared<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>(std::move(records))));
         }
 
         size_t forgotten = 0;
@@ -274,7 +274,7 @@ private:
     static constexpr size_t MemLimit = 512_KB;
     size_t MemUsage = 0;
 
-    THashMap<TActorId, TVector<IChangeRecord::TPtr>> RecordsToSend;
+    THashMap<TActorId, TVector<TChangeRecord::TPtr>> RecordsToSend;
     THashMap<TActorId, TVector<ui64>> RecordsToForget;
 
 }; // TTxRequestChangeRecords
diff --git a/ydb/core/tx/datashard/datashard_ut_change_collector.cpp b/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
index 3496a46c29e..bda3c1525d4 100644
--- a/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_change_collector.cpp
@@ -79,7 +79,7 @@ auto GetChangeRecordsWithDetails(TTestActorRuntime& runtime, const TActorId& sen
     const auto details = GetChangeRecordDetails(runtime, sender, tabletId);
     UNIT_ASSERT_VALUES_EQUAL(records.size(), details.size());
 
-    THashMap<TPathId, TVector<NChangeExchange::IChangeRecord::TPtr>> result;
+    THashMap<TPathId, TVector<TChangeRecord::TPtr>> result;
     for (size_t i = 0; i < records.size(); ++i) {
         const auto& record = records.at(i);
         const auto& detail = details.at(i);
@@ -88,7 +88,7 @@ auto GetChangeRecordsWithDetails(TTestActorRuntime& runtime, const TActorId& sen
         const auto& pathId = std::get<4>(record);
         auto it = result.find(pathId);
         if (it == result.end()) {
-            it = result.emplace(pathId, TVector<NChangeExchange::IChangeRecord::TPtr>()).first;
+            it = result.emplace(pathId, TVector<TChangeRecord::TPtr>()).first;
         }
 
         it->second.push_back(
@@ -359,7 +359,7 @@ Y_UNIT_TEST_SUITE(AsyncIndexChangeCollector) {
             UNIT_ASSERT_VALUES_EQUAL(expected.size(), actual.size());
             for (size_t i = 0; i < expected.size(); ++i) {
                 UNIT_ASSERT_VALUES_EQUAL(expected.at(i), TStructRecordBase<SK>::Parse(actual.at(i)->GetBody(), tagToName));
-                UNIT_ASSERT_VALUES_EQUAL(actual.at(i)->template Get<TChangeRecord>()->GetSchemaVersion(), entry.TableId.SchemaVersion);
+                UNIT_ASSERT_VALUES_EQUAL(actual.at(i)->GetSchemaVersion(), entry.TableId.SchemaVersion);
             }
         }
     }
@@ -763,7 +763,7 @@ Y_UNIT_TEST_SUITE(CdcStreamChangeCollector) {
             UNIT_ASSERT_VALUES_EQUAL(expected.size(), actual.size());
             for (size_t i = 0; i < expected.size(); ++i) {
                 UNIT_ASSERT_VALUES_EQUAL(expected.at(i), TStructRecordBase<SK>::Parse(actual.at(i)->GetBody(), tagToName));
-                UNIT_ASSERT_VALUES_EQUAL(actual.at(i)->Get<TChangeRecord>()->GetSchemaVersion(), entry.TableId.SchemaVersion);
+                UNIT_ASSERT_VALUES_EQUAL(actual.at(i)->GetSchemaVersion(), entry.TableId.SchemaVersion);
             }
         }
     }
diff --git a/ydb/core/tx/replication/service/json_change_record.h b/ydb/core/tx/replication/service/json_change_record.h
index 89f5076fa6c..e9012b0826f 100644
--- a/ydb/core/tx/replication/service/json_change_record.h
+++ b/ydb/core/tx/replication/service/json_change_record.h
@@ -1,6 +1,8 @@
 #pragma once
 
+#include <ydb/core/change_exchange/change_exchange.h>
 #include <ydb/core/change_exchange/change_record.h>
+#include <ydb/core/change_exchange/change_sender_resolver.h>
 #include <ydb/core/protos/tx_datashard.pb.h>
 #include <ydb/core/scheme/scheme_tablecell.h>
 #include <ydb/core/scheme_types/scheme_type_info.h>
@@ -13,6 +15,7 @@
 #include <util/generic/ptr.h>
 #include <util/generic/vector.h>
 #include <util/memory/pool.h>
+#include <util/string/join.h>
 
 namespace NKikimr::NReplication::NService {
 
@@ -48,6 +51,35 @@ public:
     TConstArrayRef<TCell> GetKey(TMemoryPool& pool) const;
     TConstArrayRef<TCell> GetKey() const;
 
+    ui64 ResolvePartitionId(NChangeExchange::IChangeSenderResolver* const resolver) const override {
+        const auto& partitions = resolver->GetPartitions();
+        Y_ABORT_UNLESS(partitions);
+        const auto& schema = resolver->GetSchema();
+        const auto streamFormat = resolver->GetStreamFormat();
+        Y_ABORT_UNLESS(streamFormat == NKikimrSchemeOp::ECdcStreamFormatJson);
+
+        // MemoryPool.Clear();
+        const auto range = TTableRange(GetKey(/* MemoryPool */));
+        Y_ABORT_UNLESS(range.Point);
+
+        const auto it = LowerBound(
+            partitions.cbegin(), partitions.cend(), true,
+            [&](const auto& partition, bool) {
+                const int compares = CompareBorders<true, false>(
+                    partition.Range->EndKeyPrefix.GetCells(), range.From,
+                    partition.Range->IsInclusive || partition.Range->IsPoint,
+                    range.InclusiveFrom || range.Point, schema
+                );
+
+                return (compares < 0);
+            }
+        );
+
+        Y_ABORT_UNLESS(it != partitions.end());
+        return it->ShardId;
+    }
+
+    using TPtr = TIntrusivePtr<TChangeRecord>;
 private:
     TString SourceId;
     NJson::TJsonValue JsonBody;
@@ -82,6 +114,32 @@ public:
 
 }
 
+namespace NKikimr {
+
+template <>
+struct TChangeRecordContainer<NReplication::NService::TChangeRecord>
+    : public TBaseChangeRecordContainer
+{
+    TChangeRecordContainer() = default;
+
+    explicit TChangeRecordContainer(TVector<NReplication::NService::TChangeRecord::TPtr>&& records)
+        : Records(std::move(records))
+    {}
+
+
+    TVector<NReplication::NService::TChangeRecord::TPtr> Records;
+
+    TString Out() override {
+        return TStringBuilder() << "[" << JoinSeq(",", Records) << "]";
+    }
+};
+
+}
+
 Y_DECLARE_OUT_SPEC(inline, NKikimr::NReplication::NService::TChangeRecord, out, value) {
     return value.Out(out);
 }
+
+Y_DECLARE_OUT_SPEC(inline, NKikimr::NReplication::NService::TChangeRecord::TPtr, out, value) {
+    return value->Out(out);
+}
diff --git a/ydb/core/tx/replication/service/table_writer.cpp b/ydb/core/tx/replication/service/table_writer.cpp
index 345c9f19c63..7e25c543e2e 100644
--- a/ydb/core/tx/replication/service/table_writer.cpp
+++ b/ydb/core/tx/replication/service/table_writer.cpp
@@ -77,9 +77,12 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter> {
         tableId.SetSchemaVersion(TableId.SchemaVersion);
 
         TString source;
-        for (auto recordPtr : ev->Get()->Records) {
+
+        auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NReplication::NService::TChangeRecord>>>(ev->Get()->Records)->Records;
+
+        for (auto recordPtr : records) {
             MemoryPool.Clear();
-            const auto& record = *recordPtr->Get<TChangeRecord>();
+            const auto& record = *recordPtr;
             record.Serialize(*event->Record.AddChanges(), MemoryPool);
 
             if (!source) {
@@ -194,8 +197,9 @@ private:
 
 class TLocalTableWriter
     : public TActor<TLocalTableWriter>
-    , public NChangeExchange::TBaseChangeSender
+    , public NChangeExchange::TBaseChangeSender<TChangeRecord>
     , public NChangeExchange::IChangeSenderResolver
+    , public NChangeExchange::ISenderFactory
     , private NSchemeCache::TSchemeCacheHelpers
 {
     TStringBuf GetLogPrefix() const {
@@ -265,8 +269,8 @@ class TLocalTableWriter
         return result;
     }
 
-    TActorId GetChangeServer() const override {
-        return SelfId();
+    void Registered(TActorSystem*, const TActorId&) override {
+        ChangeServer = SelfId();
     }
 
     void Resolve() override {
@@ -402,34 +406,13 @@ class TLocalTableWriter
         Resolving = false;
     }
 
-    IActor* CreateSender(ui64 partitionId) override {
+    IActor* CreateSender(ui64 partitionId) const override {
         return new TTablePartitionWriter(SelfId(), partitionId, TTableId(PathId, Schema->Version));
     }
 
-    ui64 GetPartitionId(NChangeExchange::IChangeRecord::TPtr record) const override {
-        Y_ABORT_UNLESS(KeyDesc);
-        Y_ABORT_UNLESS(KeyDesc->GetPartitions());
-
-        MemoryPool.Clear();
-        const auto range = TTableRange(record->Get<TChangeRecord>()->GetKey(MemoryPool));
-        Y_ABORT_UNLESS(range.Point);
-
-        TVector<TKeyDesc::TPartitionInfo>::const_iterator it = LowerBound(
-            KeyDesc->GetPartitions().begin(), KeyDesc->GetPartitions().end(), true,
-            [&](const TKeyDesc::TPartitionInfo& partition, bool) {
-                const int compares = CompareBorders<true, false>(
-                    partition.Range->EndKeyPrefix.GetCells(), range.From,
-                    partition.Range->IsInclusive || partition.Range->IsPoint,
-                    range.InclusiveFrom || range.Point, KeyDesc->KeyColumnTypes
-                );
-
-                return (compares < 0);
-            }
-        );
-
-        Y_ABORT_UNLESS(it != KeyDesc->GetPartitions().end());
-        return it->ShardId;
-    }
+    const TVector<TKeyDesc::TPartitionInfo>& GetPartitions() const override { return KeyDesc->GetPartitions(); };
+    const TVector<NScheme::TTypeInfo>& GetSchema() const override { return KeyDesc->KeyColumnTypes; }
+    NKikimrSchemeOp::ECdcStreamFormat GetStreamFormat() const override { return NKikimrSchemeOp::ECdcStreamFormatJson; }
 
     void Handle(TEvWorker::TEvData::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
@@ -455,7 +438,7 @@ class TLocalTableWriter
     void Handle(NChangeExchange::TEvChangeExchange::TEvRequestRecords::TPtr& ev) {
         LOG_D("Handle " << ev->Get()->ToString());
 
-        TVector<NChangeExchange::IChangeRecord::TPtr> records(::Reserve(ev->Get()->Records.size()));
+        TVector<NReplication::NService::TChangeRecord::TPtr> records(::Reserve(ev->Get()->Records.size()));
 
         for (const auto& record : ev->Get()->Records) {
             auto it = PendingRecords.find(record.Order);
@@ -517,7 +500,7 @@ public:
 
     explicit TLocalTableWriter(const TPathId& tablePathId)
         : TActor(&TThis::StateWork)
-        , TBaseChangeSender(this, this, tablePathId)
+        , TBaseChangeSender(this, this, this, TActorId(), tablePathId)
         , MemoryPool(256)
     {
     }
@@ -547,7 +530,7 @@ private:
     TLightweightSchema::TCPtr Schema;
     bool Resolving = false;
 
-    TMap<ui64, NChangeExchange::IChangeRecord::TPtr> PendingRecords;
+    TMap<ui64, NReplication::NService::TChangeRecord::TPtr> PendingRecords;
 
 }; // TLocalTableWriter