Что значит «вызвано: java.lang.IllegalArgumentException: смещения в порядке, найденные в списке» в Kafka?

Question

После того, как один из узлов Kafka был уничтожен, новые брокеры Kafka не могут присоединиться к кластеру, и после каждого перезапуска я вижу сообщение в его журналах:

[2019-03-13 13:52:11,673] ERROR [ReplicaFetcher replicaId=1016, leaderId=1004, fetcherId=2] Error due to (kafka.server.ReplicaFetcherThread)
kafka.common.KafkaException: Error processing data for partition log.system-11 offset 161122487
        at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:205)
        at scala.Option.foreach(Option.scala:257)
        at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6(AbstractFetcherThread.scala:172)
        at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6$adapted(AbstractFetcherThread.scala:169)
        at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:52)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
        at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$5(AbstractFetcherThread.scala:169)
        at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
        at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:217)
        at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:169)
        at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:113)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)
Caused by: java.lang.IllegalArgumentException: Out of order offsets found in List(161122487, 
...
)
        at kafka.log.Log.$anonfun$append$2(Log.scala:683)
        at kafka.log.Log.maybeHandleIOException(Log.scala:1669)
        at kafka.log.Log.append(Log.scala:624)
        at kafka.log.Log.appendAsFollower(Log.scala:607)
        at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:102)
        at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:41)
        at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:184)
        ... 12 more

После проблемы я вижу несколько сообщений вроде:

[2019-03-13 13:52:11,792] INFO [ReplicaFetcherManager on broker 1016] Removed fetcher for partitions  (kafka.server.ReplicaFetcherManager)
[2019-03-13 13:52:11,793] INFO [ReplicaFetcherManager on broker 1016] Added fetcher for partitions List() (kafka.server.ReplicaFetcherManager)

Что можно с этим сделать и можно ли это исправить? К счастью, меня не волнуют данные в этом кластере, но вся эта ситуация заставляет меня беспокоиться о возможных сбоях в работе.