class_docstr: An operator that produces messages to a Kafka topic.
  
  Registers a producer to a kafka topic and publishes messages to the log.
  
  :param kafka_config_id: The connection object to use, defaults to "kafka_default"
  :param topic: The topic the producer should produce to, defaults to None
  :param producer_function: The function that generates key/value pairs as messages for production,
      defaults to None
  :param producer_function_args: Additional arguments to be applied to the producer callable,
      defaults to None
  :param producer_function_kwargs: Additional keyword arguments to be applied to the producer callable,
      defaults to None
  :param delivery_callback: The callback to apply after delivery(or failure) of a message, defaults to None
  :param synchronous: If writes to kafka should be fully synchronous, defaults to True
  :param poll_timeout: How long of a delay should be applied when calling poll after production to kafka,
      defaults to 0
  :raises AirflowException: _description_
  
  .. seealso::
      For more information on how to use this operator, take a look at the guide:
      :ref:`howto/operator:ProduceToTopicOperator`
class_docstr_tok: ["An","operator","that","produces","messages","to","a","Kafka","topic",".","Registers","a","producer","to","a","kafka","topic","and","publishes","messages","to","the","log",".",":","param","kafka_config_id",":","The","connection","object","to","use",",","defaults","to","``","kafka_default","''",":","param","topic",":","The","topic","the","producer","should","produce","to",",","defaults","to","None",":","param","producer_function",":","The","function","that","generates","key\/value","pairs","as","messages","for","production",",","defaults","to","None",":","param","producer_function_args",":","Additional","arguments","to","be","applied","to","the","producer","callable",",","defaults","to","None",":","param","producer_function_kwargs",":","Additional","keyword","arguments","to","be","applied","to","the","producer","callable",",","defaults","to","None",":","param","delivery_callback",":","The","callback","to","apply","after","delivery","(","or","failure",")","of","a","message",",","defaults","to","None",":","param","synchronous",":","If","writes","to","kafka","should","be","fully","synchronous",",","defaults","to","True",":","param","poll_timeout",":","How","long","of","a","delay","should","be","applied","when","calling","poll","after","production","to","kafka",",","defaults","to","0",":","raises","AirflowException",":","_description_","..","seealso",":",":","For","more","information","on","how","to","use","this","operator",",","take","a","look","at","the","guide",":",":","ref",":","`","howto\/operator",":","ProduceToTopicOperator","`"]
code: def execute(self, context) -> None:
      # Get producer and callable
      producer = KafkaProducerHook(
          kafka_config_id=self.kafka_config_id
      ).get_producer()
  
      if isinstance(self.producer_function, str):
          self.producer_function = import_string(self.producer_function)
  
      producer_callable = partial(
          self.producer_function,  # type: ignore
          *self.producer_function_args,
          **self.producer_function_kwargs,
      )
  
      # For each returned k/v in the callable : publish and flush if needed.
      for k, v in producer_callable():
          producer.produce(
              self.topic,
              key=k,
              value=v,
              on_delivery=self.delivery_callback,
          )
          producer.poll(self.poll_timeout)
          if self.synchronous:
              producer.flush()
  
      producer.flush()
  
file_path: airflow/airflow/providers/apache/kafka/operators/produce.py
imports: from __future__ import annotations
  import logging
  from functools import partial
  from typing import Any, Callable, Sequence
  from airflow.exceptions import AirflowException
  from airflow.models import BaseOperator
  from airflow.providers.apache.kafka.hooks.produce import KafkaProducerHook
  from airflow.utils.module_loading import import_string
label_desc: Use image node_id 2 for calling the ProduceToTopicOperator obj's underlying member method code with example usage: obj.execute(context) without return types