Can't consume data from topic after re-installation kafka

[idir ait hammi <em...@gmail.com>]

Hello all,

As said in in the title after re-installing kafka on cloudera cluster we
have problem with consuming data on topic,  we still can product on topic
but nothing displayed on consumers.

we think the problem is coming from leader election because we have many
errors on the log brokers who says

2020-01-23 14:56:15,467 ERROR state.change.logger: [Controller id=414
epoch=101] Controller 414 epoch 101 failed to change state for
partition __consumer_offsets-5 from OfflinePartition to
OnlinePartition
kafka.common.StateChangeFailedException: Failed to elect leader for
partition __consumer_offsets-5 under strategy
OfflinePartitionLeaderElectionStrategy
	at kafka.controller.PartitionStateMachine$$anonfun$doElectLeaderForPartitions$3.apply(PartitionStateMachine.scala:390)
	at kafka.controller.PartitionStateMachine$$anonfun$doElectLeaderForPartitions$3.apply(PartitionStateMachine.scala:388)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at kafka.controller.PartitionStateMachine.doElectLeaderForPartitions(PartitionStateMachine.scala:388)
	at kafka.controller.PartitionStateMachine.electLeaderForPartitions(PartitionStateMachine.scala:315)
	at kafka.controller.PartitionStateMachine.doHandleStateChanges(PartitionStateMachine.scala:225)
	at kafka.controller.PartitionStateMachine.handleStateChanges(PartitionStateMachine.scala:141)
	at kafka.controller.PartitionStateMachine.triggerOnlinePartitionStateChange(PartitionStateMachine.scala:123)
	at kafka.controller.PartitionStateMachine.triggerOnlinePartitionStateChange(PartitionStateMachine.scala:109)
	at kafka.controller.PartitionStateMachine.startup(PartitionStateMachine.scala:66)
	at kafka.controller.KafkaController.onControllerFailover(KafkaController.scala:266)
	at kafka.controller.KafkaController.kafka$controller$KafkaController$$elect(KafkaController.scala:1271)
	at kafka.controller.KafkaController$Startup$.process(KafkaController.scala:1184)
	at kafka.controller.ControllerEventManager$ControllerEventThread$$anonfun$doWork$1.apply$mcV$sp(ControllerEventManager.scala:94)
	at kafka.controller.ControllerEventManager$ControllerEventThread$$anonfun$doWork$1.apply(ControllerEventManager.scala:94)
	at kafka.controller.ControllerEventManager$ControllerEventThread$$anonfun$doWork$1.apply(ControllerEventManager.scala:94)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:31)
	at kafka.controller.ControllerEventManager$ControllerEventThread.doWork(ControllerEventManager.scala:93)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)

if someone can help us on the subject


Thanks in advance.