code: def produce(topic, conf):
      
  
      from confluent_kafka.avro import AvroProducer
  
      producer = AvroProducer(conf, default_value_schema=record_schema)
  
      print("Producing user records to topic {}. ^c to exit.".format(topic))
      while True:
          # Instantiate new User, populate fields, produce record, execute callbacks.
          record = User()
          try:
              record.name = input("Enter name: ")
              record.favorite_number = int(input("Enter favorite number: "))
              record.favorite_color = input("Enter favorite color: ")
  
              # The message passed to the delivery callback will already be serialized.
              # To aid in debugging we provide the original object to the delivery callback.
              producer.produce(topic=topic, value=record.to_dict(),
                               callback=lambda err, msg, obj=record: on_delivery(err, msg, obj))
              # Serve on_delivery callbacks from previous asynchronous produce()
              producer.poll(0)
          except KeyboardInterrupt:
              break
          except ValueError:
              print("Invalid input, discarding record...")
              continue
  
      print("\nFlushing records...")
      producer.flush()
docs: Produce User records