我正在用python运行合流的kafka客户端。目前，我在尝试生成和消费消息时没有收到任何错误，但问题是生产者说它成功了，但是消费者找不到任何消息。
我已经创建了一个主题，这是我正在使用的类：

from confluent_kafka import Producer, Consumer
from config import config
import json

class Kafka:
    """
    Kafka Handler.
    """

    def __init__(self, kafka_brokers_sasl, api_key):
        """
        Arguments:
            kafka_brokers_sasl {str} -- String containing kafka brokers separated by comma (no spaces)
            api_key {str} -- Kafka Api Key
        """

        self.driver_options = {
            'bootstrap.servers': kafka_brokers_sasl,
            'sasl.mechanisms': 'PLAIN',
            'security.protocol': 'SASL_SSL',
            'sasl.username': 'token',
            'sasl.password': api_key,
            'log.connection.close' : False,
            #'debug': 'all'
        }

        self.producer_options = {
            'client.id': 'kafka-python-console-sample-producer'
        }
        self.producer_options.update(self.driver_options)

        self.consumer_options = {
            'client.id': 'kafka-python-console-sample-consumer',
            'group.id': 'kafka-python-console-sample-group'
        }
        self.consumer_options.update(self.driver_options)

        self.running = None

    def stop(self):
        self.running = False

    def delivery_report(self, err, msg):
        """ Called once for each message produced to indicate delivery result.
            Triggered by poll() or flush(). """
        if err is not None:
            print('Message delivery failed: {}'.format(err))
        else:
            print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

    def produce(self, topic, data): # Function for producing/uploading data to a Kafka topic

        p = Producer(self.producer_options)

        print("Running?")

        # Asynchronously produce a message, the delivery report callback will be triggered from poll() above, or flush() below, when the message has been successfully delivered or failed permanently.
        p.produce(topic, data, callback=self.delivery_report)

        # Wait for any outstanding messages to be delivered and delivery report callbacks to be triggered.
        p.flush()
        print("Done?")

    def consume(self, topic, method_class=None): # Function for consuming/reading data from a Kafka topic. Works as a listener and triggers the run() function on a method_class
        print("raaa")

        kafka_consumer = Consumer(self.consumer_options)

        kafka_consumer.subscribe([topic])

        # Now loop on the consumer to read messages
        print("Running?")
        self.running = True
        while self.running:
            msg = kafka_consumer.poll()

            print(msg)

            if msg is not None and msg.error() is None:
                print('Message consumed: topic={0}, partition={1}, offset={2}, key={3}, value={4}'.format(
                    msg.topic(),
                    msg.partition(),
                    msg.offset(),
                    msg.key().decode('utf-8'),
                    msg.value().decode('utf-8')))
            else:
                print('No messages consumed')

        print("Here?")
        kafka_consumer.unsubscribe()
        kafka_consumer.close()
        print("Ending?")

mock = {'yas': 'yas', 'yas2': 'yas2'}
kafka = Kafka(config['kafka']['kafka_brokers_sasl'], config['kafka']['api_key'])
kafka.produce(config['kafka']['topic'], json.dumps(mock))
kafka.consume(config['kafka']['topic'])

运行这个我得到了指纹：

Running?
Message delivered to DANIEL_TEST [0]
Done?
raaa
Running?
<cimpl.Message object at 0x104e4c390>
No messages consumed