[fix](memtracker) PushBrokerReader mem_pool has no tracker to control…

… memory apache#17189

#0  doris::MemTracker::consumption (this=0x0) at /root/doris/be/src/runtime/memory/mem_tracker.h:65
#1  doris::PushHandler::_convert_v2 (this=this@entry=0x7ffec1e17f90, cur_tablet=..., cur_rowset=cur_rowset@entry=0x7ffec1e177a0, tablet_schema=...)
    at /root/doris/be/src/olap/push_handler.cpp:247
#2  0x000055555a271bc6 in doris::PushHandler::_do_streaming_ingestion (this=this@entry=0x7ffec1e17f90, tablet=..., request=...,
    push_type=push_type@entry=doris::PUSH_NORMAL_V2, tablet_info_vec=tablet_info_vec@entry=0x7ffec1e185d0)
    at /var/local/ldb-toolchain/include/c++/11/ext/atomicity.h:109
#3  0x000055555a27222c in doris::PushHandler::process_streaming_ingestion

be/src/olap/push_handler.cpp

@@ -819,8 +819,10 @@ Status PushBrokerReader::init(const Schema* schema, const TBrokerScanRange& t_sc
     }
     _runtime_profile = _runtime_state->runtime_profile();
     _runtime_profile->set_name("PushBrokerReader");
-    _mem_pool.reset(new MemPool());
-    _tuple_buffer_pool.reset(new MemPool());
+    _mem_tracker = std::make_unique<MemTracker>(
+            fmt::format("PushBrokerReader#InstanceId={}", print_id(params.fragment_instance_id)));
+    _mem_pool.reset(new MemPool(_mem_tracker.get()));
+    _tuple_buffer_pool.reset(new MemPool(_mem_tracker.get()));
 
     _counter.reset(new ScannerCounter());
 

be/src/olap/push_handler.h

@@ -205,6 +205,7 @@ class PushBrokerReader {
     const Schema* _schema;
     std::unique_ptr<RuntimeState> _runtime_state;
     RuntimeProfile* _runtime_profile;
+    std::unique_ptr<MemTracker> _mem_tracker;
     std::unique_ptr<MemPool> _mem_pool;
     std::unique_ptr<MemPool> _tuple_buffer_pool;
     std::unique_ptr<ScannerCounter> _counter;