[fix](sink) Fix NodeChannel add_block_closure null pointer

[xinyiZzz]

Proposed changes

NodeChannel add_block_closure null pointer when canceled before open_wait new closure.

(gdb) bt
#0  std::__atomic_base<bool>::load (this=0x380, __m=std::memory_order::seq_cst)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/atomic_base.h:481
#1  std::atomic<bool>::operator bool (this=0x380) at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/atomic:87
#2  doris::stream_load::ReusableClosure<doris::PTabletWriterAddBlockResult>::is_packet_in_flight (this=0x0)
    at /root/repo/doris/be/src/vec/sink/vtablet_sink.h:165
#3  doris::stream_load::VNodeChannel::is_rpc_done (this=0x7fbc1b223510) at /root/repo/doris/be/src/vec/sink/vtablet_sink.cpp:908
#4  doris::stream_load::VOlapTableSink::is_close_done()::$_0::operator()(std::shared_ptr<doris::stream_load::VNodeChannel> const&) const (
    this=0x7fbfb9c96550, ch=...) at /root/repo/doris/be/src/vec/sink/vtablet_sink.cpp:1543
#5  std::__invoke_impl<void, doris::stream_load::VOlapTableSink::is_close_done()::$_0&, std::shared_ptr<doris::stream_load::VNodeChannel> const&>(std::__invoke_other, doris::stream_load::VOlapTableSink::is_close_done()::$_0&, std::shared_ptr<doris::stream_load::VNodeChannel> const&) (__f=..., __args=...)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/invoke.h:61
#6  std::__invoke_r<void, doris::stream_load::VOlapTableSink::is_close_done()::$_0&, std::shared_ptr<doris::stream_load::VNodeChannel> const&>(doris::stream_load::VOlapTableSink::is_close_done()::$_0&, std::shared_ptr<doris::stream_load::VNodeChannel> const&) (__fn=..., __args=...)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/invoke.h:111
#7  std::_Function_handler<void (std::shared_ptr<doris::stream_load::VNodeChannel> const&), doris::stream_load::VOlapTableSink::is_close_done()::$_0>::_M_invoke(std::_Any_data const&, std::shared_ptr<doris::stream_load::VNodeChannel> const&) (__functor=..., __args=...)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_function.h:291
#8  0x000055d14322c67a in std::function<void (std::shared_ptr<doris::stream_load::VNodeChannel> const&)>::operator()(std::shared_ptr<doris::stream_load::VNodeChannel> const&) const (this=<optimized out>, __args=...)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_function.h:560
#9  doris::stream_load::IndexChannel::for_each_node_channel(std::function<void (std::shared_ptr<doris::stream_load::VNodeChannel> const&)> const&) (
    this=<optimized out>, func=...) at /root/repo/doris/be/src/vec/sink/vtablet_sink.h:408
#10 doris::stream_load::VOlapTableSink::is_close_done (this=<optimized out>) at /root/repo/doris/be/src/vec/sink/vtablet_sink.cpp:1541
#11 0x000055d143255c1d in doris::pipeline::DataSinkOperator<doris::pipeline::OlapTableSinkOperatorBuilder>::is_pending_finish (this=<optimized out>)
    at /root/repo/doris/be/src/pipeline/exec/operator.h:292
#12 0x000055d143296f19 in doris::pipeline::PipelineTask::is_pending_finish (this=0x7fbd1bb3ae00) at /root/repo/doris/be/src/pipeline/pipeline_task.h:137
#13 doris::pipeline::TaskScheduler::_try_close_task (this=0x7fc02c3fabe0, task=0x7fbd1bb3ae00, state=doris::pipeline::PipelineTaskState::CANCELED)
    at /root/repo/doris/be/src/pipeline/task_scheduler.cpp:330
#14 0x000055d143296870 in doris::pipeline::TaskScheduler::_do_work (this=0x7fc02c3fabe0, index=1)
    at /root/repo/doris/be/src/pipeline/task_scheduler.cpp:250
#15 0x000055d13cd24b0f in doris::ThreadPool::dispatch_thread (this=0x7fc030b01e00) at /root/repo/doris/be/src/util/threadpool.cpp:531
#16 0x000055d13cd1a9ac in std::function<void ()>::operator()() const (this=0x7fbfb9c96550)
    at /var/local/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_function.h:560
#17 doris::Thread::supervise_thread (arg=0x7fc02ec3d340) at /root/repo/doris/be/src/util/thread.cpp:465
#18 0x00007fc05523a609 in start_thread (arg=<optimized out>) at pthread_create.c:477
#19 0x00007fc0554c9163 in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 

Further comments

If this is a relatively large or complex change, kick off the discussion at dev@doris.apache.org by explaining why you chose the solution you did and what alternatives you considered, etc...

[github-actions]

clang-tidy review says "All clean, LGTM! 👍"

[xinyiZzz]

run buildall

[xinyiZzz]

run buildall

[github-actions]

clang-tidy review says "All clean, LGTM! 👍"

[github-actions]

PR approved by at least one committer and no changes requested.

[github-actions]

PR approved by anyone and no changes requested.

[Gabriel39]

LGTM