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 execute_complete(self, context, event=None):
      self.event_triggered_function(event, **context)
  
      self.defer(
          trigger=AwaitMessageTrigger(
              topics=self.topics,
              apply_function=self.apply_function,
              apply_function_args=self.apply_function_args,
              apply_function_kwargs=self.apply_function_kwargs,
              kafka_config_id=self.kafka_config_id,
              poll_timeout=self.poll_timeout,
              poll_interval=self.poll_interval,
          ),
          method_name="execute_complete",
      )
  
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 3 for calling the AwaitMessageTriggerFunctionSensor obj's underlying member method code with example usage: obj.execute_complete(context, event) without return types