junrao commented on a change in pull request #9596:
URL: https://github.com/apache/kafka/pull/9596#discussion_r524673732
##########
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:
Hmm, since we are about to shut down the JVM, should we just log a WARN
here instead of throwing the exception?
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -83,6 +87,51 @@ class LogManagerTest {
log.appendAsLeader(TestUtils.singletonRecords("test".getBytes()),
leaderEpoch = 0)
}
+ /**
+ * Tests that all internal futures are completed before
LogManager.shutdown() returns to the
+ * caller during error situations.
+ */
+ @Test
+ def testHandlingExceptionsDuringShutdown(): Unit = {
+ logManager.shutdown()
Review comment:
Hmm, do we need this given that we do this in tearDown() already?
----------------------------------------------------------------
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
