MultiThreadExecutor may throw exception after destroying subscription with non-default callback group

[KKSTB]

Bug report

Required Info:

• Operating System:
  • Ubuntu 22.04
• Installation type:
  • binaries
• Version or commit hash:
  • ros-iron-rclpy/jammy,now 4.1.4-1jammy.20231117.092234 amd64
• DDS implementation:
  • Cyclone DDS
• Client library (if applicable):
  • rclpy

Steps to reproduce issue

1. Setup publisher on a different program and repeatedly publishes data
2. In the test program, create MultiThreadExecutor and create a thread to spin it
3. Repeatedly create and destroy subscription with new MutuallyExclusiveCallbackGroup like this:

for index in range(100):
    test_sub = rclpy_node.create_subscription(std_msgs.msg.String, '/test', self.testCB, qos_profile=1, callback_group=rclpy.callback_groups.MutuallyExclusiveCallbackGroup())
    time.sleep(0.04)  # Wait for some messages
    rclpy_node.destroy_subscription(test_sub)

Expected behavior

Succeed without exception

Actual behavior

Traceback (most recent call last):
  File "/home/test/test.py", line 41, in _spin_rclpy_executor
    self.executor.spin()
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/executors.py", line 295, in spin
    self.spin_once()
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/executors.py", line 813, in spin_once
    self._spin_once_impl(timeout_sec)
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/executors.py", line 810, in _spin_once_impl
    future.result()
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/task.py", line 94, in result
    raise self.exception()
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/task.py", line 239, in __call__
    self._handler.send(None)
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/executors.py", line 436, in handler
    arg = take_from_wait_list(entity)
  File "/opt/ros/iron/lib/python3.10/site-packages/rclpy/executors.py", line 356, in _take_subscription
    with sub.handle:
rclpy._rclpy_pybind11.InvalidHandle: cannot use Destroyable because destruction was requested

Additional information

Exception seems doesn't happen if the subscription is in the default callback group.

This problem also affects node.destroy_node() of an already removed node, e.g. like this:

while rclpy.ok():
    executor.spin_once()
    if not self._destroy_node_queue.empty():
        rclpy_node = self._destroy_node_queue.get()
        executor.remove_node(rclpy_node)
        rclpy_node.destroy_node()
        self._destroy_node_queue.task_done()