title: kafka with flask. KafkaTimeoutError: Failed to update metadata after 60.0 secs
body: <p>How to integrate kafka with flask? I have done it but facing error while sending message (<code>producer_instance.send(topic_name, key=key_bytes, value=value_bytes)</code>). It says <code>KafkaTimeoutError: Failed to update metadata after 60.0 secs.</code>. I'm publishing my codes here.</p>
  
  <pre><code>    #views.py
      from app import app
      from db.mongo import Database
      from flask import jsonify
      import json
      from app.kafka import Producer
  
      @app.route('/')
      def index():
          kafka_producer = None
          results = get_data()
          if len(results) &gt; 0:
              kafka_producer = Producer.connect_kafka_producer()
              Producer.publish_message(kafka_producer, 'raw_country_info', 'raw', str(results).strip())
          if kafka_producer is not None:
              kafka_producer.close()
          return json.dumps(results, default=str)
  
      def get_data():
          database = Database().get_db()
          results = []
          for doc in database["database"].get_collection('country_info').find().limit(10):
              results.append(doc)
          return results
  
      #Producer.py
      import json
      from time import sleep
      from bs4 import BeautifulSoup
      from kafka import KafkaConsumer, KafkaProducer
  
      def connect_kafka_producer():
          _producer = None
          try:
              _producer = KafkaProducer(bootstrap_servers=['localhost:9092'], api_version=(0, 10))
          except Exception as ex:
              print('Exception while connecting Kafka')
              print(str(ex))
          finally:
              return _producer
  
      def publish_message(producer_instance, topic_name, key, value):
          try:
              key_bytes = bytes(key, encoding='utf-8')
              value_bytes = bytes(value, encoding='utf-8')
              print("---------", key_bytes, value_bytes)
              producer_instance.send(topic_name, key=key_bytes, value=value_bytes)
              producer_instance.flush()
              print('Message published successfully.')
          except Exception as ex:
              print('Exception in publishing message')
              print(str(ex))
  </code></pre>