(kafka) branch trunk updated: KAFKA-16363 Storage tool crashes if dir is unavailable (#15733)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 44196984bd2 KAFKA-16363 Storage tool crashes if dir is unavailable (#15733)
44196984bd2 is described below

commit 44196984bd244b1e1cc1a3bcda85035d3dee2dc8
Author: Michael Westerby <mi...@apple.com>
AuthorDate: Wed Apr 17 12:03:51 2024 +0100

    KAFKA-16363 Storage tool crashes if dir is unavailable (#15733)
    
    Reviewers: Igor Soarez <so...@apple.com>, Federico Valeri <fe...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
 core/src/main/scala/kafka/tools/StorageTool.scala  |  8 +++-
 .../scala/unit/kafka/tools/StorageToolTest.scala   | 53 ++++++++++++++++++++++
 2 files changed, 59 insertions(+), 2 deletions(-)

diff --git a/core/src/main/scala/kafka/tools/StorageTool.scala b/core/src/main/scala/kafka/tools/StorageTool.scala
index 6f0fad731eb..148be404bf7 100644
--- a/core/src/main/scala/kafka/tools/StorageTool.scala
+++ b/core/src/main/scala/kafka/tools/StorageTool.scala
@@ -441,8 +441,12 @@ object StorageTool extends Logging {
         "Use --ignore-formatted to ignore this directory and format the others.")
     }
     if (!copier.errorLogDirs().isEmpty) {
-      val firstLogDir = copier.errorLogDirs().iterator().next()
-      throw new TerseFailure(s"I/O error trying to read log directory $firstLogDir.")
+      copier.errorLogDirs().forEach(errorLogDir => {
+        stream.println(s"I/O error trying to read log directory $errorLogDir. Ignoring...")
+      })
+      if (metaPropertiesEnsemble.emptyLogDirs().isEmpty && copier.logDirProps().isEmpty) {
+        throw new TerseFailure("No available log directories to format.")
+      }
     }
     if (metaPropertiesEnsemble.emptyLogDirs().isEmpty) {
       stream.println("All of the log directories are already formatted.")
diff --git a/core/src/test/scala/unit/kafka/tools/StorageToolTest.scala b/core/src/test/scala/unit/kafka/tools/StorageToolTest.scala
index 4470d877536..a44ac0c63ad 100644
--- a/core/src/test/scala/unit/kafka/tools/StorageToolTest.scala
+++ b/core/src/test/scala/unit/kafka/tools/StorageToolTest.scala
@@ -192,6 +192,59 @@ Found problem:
     } finally Utils.delete(tempDir)
   }
 
+  private def runFormatCommand(stream: ByteArrayOutputStream, directories: Seq[String], ignoreFormatted: Boolean = false): Int = {
+    val metaProperties = new MetaProperties.Builder().
+      setVersion(MetaPropertiesVersion.V1).
+      setClusterId("XcZZOzUqS4yHOjhMQB6JLQ").
+      setNodeId(2).
+      build()
+    val bootstrapMetadata = StorageTool.buildBootstrapMetadata(MetadataVersion.latestTesting(), None, "test format command")
+    StorageTool.formatCommand(new PrintStream(stream), directories, metaProperties, bootstrapMetadata, MetadataVersion.latestTesting(), ignoreFormatted)
+  }
+
+  @Test
+  def testFormatSucceedsIfAllDirectoriesAreAvailable(): Unit = {
+    val availableDir1 = TestUtils.tempDir()
+    val availableDir2 = TestUtils.tempDir()
+    val stream = new ByteArrayOutputStream()
+    assertEquals(0, runFormatCommand(stream, Seq(availableDir1.toString, availableDir2.toString)))
+    assertTrue(stream.toString().contains("Formatting %s".format(availableDir1)))
+    assertTrue(stream.toString().contains("Formatting %s".format(availableDir2)))
+  }
+
+  @Test
+  def testFormatSucceedsIfAtLeastOneDirectoryIsAvailable(): Unit = {
+    val availableDir1 = TestUtils.tempDir()
+    val unavailableDir1 = TestUtils.tempFile()
+    val stream = new ByteArrayOutputStream()
+    assertEquals(0, runFormatCommand(stream, Seq(availableDir1.toString, unavailableDir1.toString)))
+    assertTrue(stream.toString().contains("I/O error trying to read log directory %s. Ignoring...".format(unavailableDir1)))
+    assertTrue(stream.toString().contains("Formatting %s".format(availableDir1)))
+    assertFalse(stream.toString().contains("Formatting %s".format(unavailableDir1)))
+  }
+
+  @Test
+  def testFormatFailsIfAllDirectoriesAreUnavailable(): Unit = {
+    val unavailableDir1 = TestUtils.tempFile()
+    val unavailableDir2 = TestUtils.tempFile()
+    val stream = new ByteArrayOutputStream()
+    assertEquals("No available log directories to format.", assertThrows(classOf[TerseFailure],
+      () => runFormatCommand(stream, Seq(unavailableDir1.toString, unavailableDir2.toString))).getMessage)
+    assertTrue(stream.toString().contains("I/O error trying to read log directory %s. Ignoring...".format(unavailableDir1)))
+    assertTrue(stream.toString().contains("I/O error trying to read log directory %s. Ignoring...".format(unavailableDir2)))
+  }
+
+  @Test
+  def testFormatSucceedsIfAtLeastOneFormattedDirectoryIsAvailable(): Unit = {
+    val availableDir1 = TestUtils.tempDir()
+    val stream = new ByteArrayOutputStream()
+    assertEquals(0, runFormatCommand(stream, Seq(availableDir1.toString)))
+
+    val stream2 = new ByteArrayOutputStream()
+    val unavailableDir1 = TestUtils.tempFile()
+    assertEquals(0, runFormatCommand(stream2, Seq(availableDir1.toString, unavailableDir1.toString), ignoreFormatted = true))
+  }
+
   @Test
   def testFormatWithInvalidClusterId(): Unit = {
     val config = new KafkaConfig(newSelfManagedProperties())