class_docstr: Defer until a specific message is published to Kafka, trigger a registered function, then resume waiting.
  
  The behavior of the consumer for this trigger is as follows:
  - poll the Kafka topics for a message
  - if no message returned, sleep
  - process the message with provided callable and commit the message offset
  - if callable returns any data, raise a TriggerEvent with the return data
  - else continue to next message
  - return event (as default xcom or specific xcom key)
  
  
  :param kafka_config_id: The connection object to use, defaults to "kafka_default"
  :param topics: Topics (or topic regex) to use for reading from
  :param apply_function: The function to apply to messages to determine if an event occurred. As a dot
      notation string.
  :param event_triggered_function: The callable to trigger once the apply_function encounters a
      positive event.
  :param apply_function_args: Arguments to be applied to the processing function, defaults to None
  :param apply_function_kwargs: Key word arguments to be applied to the processing function,
      defaults to None
  :param poll_timeout: How long the kafka consumer should wait for a message to arrive from the kafka
      cluster, defaults to 1
  :param poll_interval: How long the kafka consumer should sleep after reaching the end of the Kafka log,
      defaults to 5
code: def __init__(
      self,
      topics: Sequence[str],
      apply_function: str,
      event_triggered_function: Callable,
      kafka_config_id: str = "kafka_default",
      apply_function_args: Sequence[Any] | None = None,
      apply_function_kwargs: dict[Any, Any] | None = None,
      poll_timeout: float = 1,
      poll_interval: float = 5,
      **kwargs: Any,
  ) -> None:
      super().__init__(**kwargs)
  
      self.topics = topics
      self.apply_function = apply_function
      self.apply_function_args = apply_function_args
      self.apply_function_kwargs = apply_function_kwargs
      self.kafka_config_id = kafka_config_id
      self.poll_timeout = poll_timeout
      self.poll_interval = poll_interval
      self.event_triggered_function = event_triggered_function
  
      if not callable(self.event_triggered_function):
          raise TypeError(
              "parameter event_triggered_function is expected to be of type callable,"
              f"got {type(event_triggered_function)}"
          )
  
file_path: airflow/airflow/providers/apache/kafka/sensors/kafka.py
imports: from __future__ import annotations
  from typing import Any, Callable, Sequence
  from airflow.models import BaseOperator
  from airflow.providers.apache.kafka.triggers.await_message import AwaitMessageTrigger
label_desc: Use image node_id 1 to create a new AwaitMessageTriggerFunctionSensor object from inherited base classes: BaseOperator with example: obj = AwaitMessageTriggerFunctionSensor(topics, apply_function, event_triggered_function, kafka_config_id, apply_function_args, apply_function_kwargs, poll_timeout, poll_interval)