class_docstr: Test Sensors
code: def test_await_message_trigger_event(self):
      sensor = AwaitMessageTriggerFunctionSensor(
          kafka_config_id="kafka_d",
          topics=["test"],
          task_id="test",
          apply_function=_return_true,
          event_triggered_function=_return_true,
      )
  
      # task should immediately come out of deferred
      with pytest.raises(TaskDeferred):
          sensor.execute(context={})
  
file_path: airflow/tests/providers/apache/kafka/sensors/test_kafka.py
imports: from __future__ import annotations
  import json
  import logging
  import pytest
  from airflow.exceptions import TaskDeferred
  from airflow.models import Connection
  from airflow.providers.apache.kafka.sensors.kafka import AwaitMessageSensor, AwaitMessageTriggerFunctionSensor
  from airflow.utils import db
label_desc: Use image node_id 4 for calling the TestSensors obj's underlying member method code with example usage: obj.test_await_message_trigger_event() without return types