diff --git a/rosbag2_py/src/rosbag2_py/_transport.cpp b/rosbag2_py/src/rosbag2_py/_transport.cpp
index bbfa21b73a..52ef47c28c 100644
--- a/rosbag2_py/src/rosbag2_py/_transport.cpp
+++ b/rosbag2_py/src/rosbag2_py/_transport.cpp
@@ -141,22 +141,27 @@ class Player
 
 class Recorder
 {
-private:
-  std::unique_ptr<rclcpp::executors::SingleThreadedExecutor> exec_;
-
 public:
   Recorder()
   {
-    rclcpp::init(0, nullptr);
-    exec_ = std::make_unique<rclcpp::executors::SingleThreadedExecutor>();
+    auto init_options = rclcpp::InitOptions();
+    init_options.shutdown_on_signal = false;
+    rclcpp::init(0, nullptr, init_options, rclcpp::SignalHandlerOptions::SigInt);
+    rclcpp::uninstall_signal_handlers();
+
     std::signal(
       SIGTERM, [](int /* signal */) {
-        rclcpp::shutdown();
+        rosbag2_py::Recorder::cancel();
+      });
+    std::signal(
+      SIGINT, [](int /* signal */) {
+        rosbag2_py::Recorder::cancel();
       });
   }
 
   virtual ~Recorder()
   {
+//    rclcpp::install_signal_handlers(rclcpp::SignalHandlerOptions::SigTerm);
     rclcpp::shutdown();
   }
 
@@ -164,6 +169,8 @@ class Recorder
     const rosbag2_storage::StorageOptions & storage_options,
     RecordOptions & record_options)
   {
+    exit_ = false;
+    auto exec = std::make_unique<rclcpp::executors::SingleThreadedExecutor>();
     if (record_options.rmw_serialization_format.empty()) {
       record_options.rmw_serialization_format = std::string(rmw_get_serialization_format());
     }
@@ -173,20 +180,42 @@ class Recorder
       std::move(writer), storage_options, record_options);
     recorder->record();
 
-    exec_->add_node(recorder);
-    // Release the GIL for long-running record, so that calling Python code can use other threads
+    exec->add_node(recorder);
+    // Run exec->spin() in a separate thread, because we need to call exec->cancel() after
+    // recorder->stop() to be able to send notifications about bag split and close.
+    auto spin_thread = std::thread(
+      [&exec]() {
+        exec->spin();
+      });
     {
+      // Release the GIL for long-running record, so that calling Python code can use other threads
       py::gil_scoped_release release;
-      exec_->spin();
+      std::unique_lock<std::mutex> lock(wait_for_exit_mutex_);
+      wait_for_exit_cv_.wait(lock, [] {return rosbag2_py::Recorder::exit_.load();});
+      recorder->stop();
+    }
+    exec->cancel();
+    if (spin_thread.joinable()) {
+      spin_thread.join();
     }
+    exec->remove_node(recorder);
   }
 
-  void cancel()
+  static void cancel()
   {
-    exec_->cancel();
+    exit_ = true;
+    wait_for_exit_cv_.notify_all();
   }
+
+protected:
+  static std::atomic_bool exit_;
+  static std::condition_variable wait_for_exit_cv_;
+  std::mutex wait_for_exit_mutex_;
 };
 
+std::atomic_bool Recorder::exit_{false};
+std::condition_variable Recorder::wait_for_exit_cv_{};
+
 // Simple wrapper to read the output config YAML into structs
 void bag_rewrite(
   const std::vector<rosbag2_storage::StorageOptions> & input_options,
@@ -283,8 +312,9 @@ PYBIND11_MODULE(_transport, m) {
 
   py::class_<rosbag2_py::Recorder>(m, "Recorder")
   .def(py::init())
-  .def("record", &rosbag2_py::Recorder::record)
-  .def("cancel", &rosbag2_py::Recorder::cancel)
+  .def(
+    "record", &rosbag2_py::Recorder::record, py::arg("storage_options"), py::arg("record_options"))
+  .def_static("cancel", &rosbag2_py::Recorder::cancel)
   ;
 
   m.def(
diff --git a/rosbag2_py/test/test_transport.py b/rosbag2_py/test/test_transport.py
index 176d7e9fe2..d5d3f01a2d 100644
--- a/rosbag2_py/test/test_transport.py
+++ b/rosbag2_py/test/test_transport.py
@@ -85,3 +85,4 @@ def test_record_cancel(tmp_path):
     db3_path = Path(bag_path) / 'test_record_cancel_0.db3'
     assert wait_for(lambda: metadata_path.is_file() and db3_path.is_file(),
                     timeout=rclpy.duration.Duration(seconds=3))
+    record_thread.join()