code: def on_task_instance_success(
      self, previous_state, task_instance, session
  ):
      read = task_instance.xcom_pull(
          task_ids=self.task_id, key="listener"
      )
      self.write("on_task_instance_success")
      self.write(read)
  
imports: from __future__ import annotations
  from airflow.listeners import hookimpl
label_desc: Use image node_id 4 for calling the XComListener obj's underlying member method code with example usage: obj.on_task_instance_success(previous_state, task_instance, session) without return types