cmccabe commented on code in PR #13140:
URL: https://github.com/apache/kafka/pull/13140#discussion_r1083120749
##########
core/src/main/scala/kafka/raft/RaftManager.scala:
##########
@@ -47,19 +47,27 @@ import org.apache.kafka.raft.RaftConfig.{AddressSpec,
InetAddressSpec, NON_ROUTA
import org.apache.kafka.raft.{FileBasedStateStore, KafkaRaftClient,
LeaderAndEpoch, RaftClient, RaftConfig, RaftRequest, ReplicatedLog}
import org.apache.kafka.server.common.serialization.RecordSerde
import org.apache.kafka.server.util.KafkaScheduler
+import org.apache.kafka.server.fault.FaultHandler
import scala.jdk.CollectionConverters._
object KafkaRaftManager {
class RaftIoThread(
client: KafkaRaftClient[_],
- threadNamePrefix: String
+ threadNamePrefix: String,
+ fatalFaultHandler: FaultHandler
) extends ShutdownableThread(
name = threadNamePrefix + "-io-thread",
isInterruptible = false
) {
override def doWork(): Unit = {
- client.poll()
+ try {
+ client.poll()
+ } catch {
+ case t: Throwable =>
+ fatalFaultHandler.handleFault("Unexpected error in raft IO thread",
t)
Review Comment:
suggest doing `throw fatalFaultHandler.handleFault(...)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
