kowshik commented on a change in pull request #9596:
URL: https://github.com/apache/kafka/pull/9596#discussion_r524847888
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -477,27 +477,41 @@ class LogManager(logDirs: Seq[File],
jobs(dir) = jobsForDir.map(pool.submit).toSeq
}
+ var firstExceptionOpt: Option[Throwable] = Option.empty
try {
for ((dir, dirJobs) <- jobs) {
- dirJobs.foreach(_.get)
+ val errorsForDirJobs = dirJobs.map {
+ future =>
+ try {
+ future.get
+ Option.empty
+ } catch {
+ case e: ExecutionException =>
+ error(s"There was an error in one of the threads during
LogManager shutdown: ${e.getCause}")
+ Some(e.getCause)
+ }
+ }.filter{ e => e.isDefined }.map{ e => e.get }
+
+ if (firstExceptionOpt.isEmpty) {
+ firstExceptionOpt = errorsForDirJobs.headOption
+ }
- val logs = logsInDir(localLogsByDir, dir)
+ if (errorsForDirJobs.isEmpty) {
+ val logs = logsInDir(localLogsByDir, dir)
- // update the last flush point
- debug(s"Updating recovery points at $dir")
- checkpointRecoveryOffsetsInDir(dir, logs)
+ // update the last flush point
+ debug(s"Updating recovery points at $dir")
+ checkpointRecoveryOffsetsInDir(dir, logs)
- debug(s"Updating log start offsets at $dir")
- checkpointLogStartOffsetsInDir(dir, logs)
+ debug(s"Updating log start offsets at $dir")
+ checkpointLogStartOffsetsInDir(dir, logs)
- // mark that the shutdown was clean by creating marker file
- debug(s"Writing clean shutdown marker at $dir")
- CoreUtils.swallow(Files.createFile(new File(dir,
Log.CleanShutdownFile).toPath), this)
+ // mark that the shutdown was clean by creating marker file
+ debug(s"Writing clean shutdown marker at $dir")
+ CoreUtils.swallow(Files.createFile(new File(dir,
Log.CleanShutdownFile).toPath), this)
+ }
}
- } catch {
- case e: ExecutionException =>
- error(s"There was an error in one of the threads during LogManager
shutdown: ${e.getCause}")
- throw e.getCause
+ firstExceptionOpt.foreach{ e => throw e}
Review comment:
Great point. I've changed the code to do the same.
My understanding is that the exception swallow safety net exists inside
`KafkaServer.shutdown()` today, but it makes sense to also just log a warning
here instead instead of relying on the safety net:
https://github.com/apache/kafka/blob/bb34c5c8cc32d1b769a34329e34b83cda040aafc/core/src/main/scala/kafka/server/KafkaServer.scala#L732.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
