[BUG] Following a failed attempt to open the partition, it can no longer be reopened successfully on the same broker.

[superhx]

[2024-03-30 09:39:23,942] ERROR [Broker id=2] Transitioning partition(s) fail: LocalReplicaChanges(deletes = [], leaders = {perf-0=PartitionInfo(topicId = qW8-MDukSQqyQUX-QsgyEA, partition = PartitionRegistration(replicas=[2], isr=[2], removingReplicas=[], addingReplicas=[], leader=2, leaderRecoveryState=RECOVERED, leaderEpoch=0, partitionEpoch=0))}, followers = {}) (state.change.logger)
org.apache.kafka.common.errors.KafkaStorageException: failed to open ElasticUnifiedLog perf-0 in dir /tmp/kraft-broker-logs/perf-0
Caused by: java.io.IOException: mock io exception
	at kafka.log.streamaspect.ElasticLog$.apply(ElasticLog.scala:474)
	at kafka.log.streamaspect.ElasticLogManager$$anon$1.run(ElasticLogManager.scala:65)
	at kafka.log.streamaspect.utils.ExceptionUtil.maybeRecordThrowableAndRethrow(ExceptionUtil.java:27)
	at kafka.log.streamaspect.ElasticLogManager.getOrCreateLog(ElasticLogManager.scala:67)
	at kafka.log.streamaspect.ElasticLogManager$.getOrCreateLog(ElasticLogManager.scala:189)
	at kafka.log.UnifiedLog$.$anonfun$applyElasticUnifiedLog$2(UnifiedLog.scala:1916)
	at kafka.log.LocalLog$.maybeHandleIOException(LocalLog.scala:820)
	at kafka.log.UnifiedLog$.applyElasticUnifiedLog(UnifiedLog.scala:1914)
	at kafka.log.UnifiedLog$.apply(UnifiedLog.scala:1860)
	at kafka.log.LogManager.$anonfun$getOrCreateLog$1(LogManager.scala:1059)
	at scala.Option.getOrElse(Option.scala:201)
	at kafka.log.LogManager.getOrCreateLog(LogManager.scala:1007)
	at kafka.cluster.Partition.createLog(Partition.scala:399)
	at kafka.cluster.Partition.maybeCreate$1(Partition.scala:374)
	at kafka.cluster.Partition.createLogIfNotExists(Partition.scala:381)
	at kafka.cluster.Partition.$anonfun$makeLeader$1(Partition.scala:687)
	at kafka.cluster.Partition.makeLeader(Partition.scala:643)
	at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$3(ReplicaManager.scala:2830)
	at kafka.server.ReplicaManager.getOrCreatePartition(ReplicaManager.scala:2676)
	at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$2(ReplicaManager.scala:2828)
	at kafka.server.ReplicaManager.$anonfun$applyLocalLeadersDelta$2$adapted(ReplicaManager.scala:2827)
	at kafka.utils.Implicits$MapExtensionMethods$.$anonfun$forKeyValue$1(Implicits.scala:62)
	at scala.collection.mutable.HashMap$Node.foreachEntry(HashMap.scala:633)
	at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:499)
	at kafka.server.ReplicaManager.applyLocalLeadersDelta(ReplicaManager.scala:2827)
	at kafka.server.ReplicaManager.$anonfun$asyncApplyDelta$7(ReplicaManager.scala:2738)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
[2024-03-30 09:42:09,131] INFO [Broker id=2] Deleting 1 partition(s). (state.change.logger)
[2024-03-30 09:42:59,588] INFO [Broker id=2] Transitioning partition(s) info: LocalReplicaChanges(deletes = [], leaders = {}, followers = {perf-0=PartitionInfo(topicId = qW8-MDukSQqyQUX-QsgyEA, partition = PartitionRegistration(replicas=[2], isr=[2], removingReplicas=[], addingReplicas=[], leader=-1, leaderRecoveryState=RECOVERED, leaderEpoch=3, partitionEpoch=3))}) (state.change.logger)
[2024-03-30 09:43:00,200] INFO [Broker id=2] Transitioning partition(s) info: LocalReplicaChanges(deletes = [], leaders = {perf-0=PartitionInfo(topicId = qW8-MDukSQqyQUX-QsgyEA, partition = PartitionRegistration(replicas=[2], isr=[2], removingReplicas=[], addingReplicas=[], leader=2, leaderRecoveryState=RECOVERED, leaderEpoch=4, partitionEpoch=4))}, followers = {}) (state.change.logger)
[2024-03-30 09:43:00,201] INFO [Broker id=2] Transitioning 1 partition(s) to local leaders. (state.change.logger)
[2024-03-30 09:43:00,201] INFO [Broker id=2] Creating new partition perf-0 with topic id qW8-MDukSQqyQUX-QsgyEA. (state.change.logger)
[2024-03-30 09:43:00,205] ERROR [Broker id=2] Transitioning partition(s) fail: LocalReplicaChanges(deletes = [], leaders = {perf-0=PartitionInfo(topicId = qW8-MDukSQqyQUX-QsgyEA, partition = PartitionRegistration(replicas=[2], isr=[2], removingReplicas=[], addingReplicas=[], leader=2, leaderRecoveryState=RECOVERED, leaderEpoch=4, partitionEpoch=4))}, followers = {}) (state.change.logger)
org.apache.kafka.common.errors.KafkaStorageException: The log dir /tmp/kraft-broker-logs/perf-0 is already offline due to a previous IO exception.