我有一个cloudera集群和一个集群化的kafka服务。我有两个Kafka控制器的例子，比如c1和c2。
当c1是主动控制器时，一切似乎都正常工作。由于某种原因，c2成为主动控制器。通过kafka console producer发送的一些消息没有被kafka console consumer接收（正好有一半发送的消息没有被consumer接收，每两条消息中就有一条消息）。我不确定这是否是由于活动控制器的变化和某些分区无法读取。
另外，如果我关闭c1，只保持c2运行，我还有另一个问题。如果我尝试从kafka开始一个先前工作的streamsets流读取，我会得到错误消息“cannotretrievemetadatafortopicx”，似乎主题元数据只存在于c1中，在这个场景中，c1是离线的。
如果在关闭第一个代理后打开kafka控制台使用者，则会出现以下异常：

WARN [console-consumer-16627_node10.agatha-cluster-1515508696963-2e45e6d8-leader-finder-thread]: 
Failed to find leader for Set(testD-1, testD-0) 
(kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(testD)] 
from broker [ArrayBuffer(BrokerEndPoint(183,110.250.17.242,9092))] failed

当我试图和几个经纪人一起使用Kafka的时候，我做错了什么？