[GitHub] [kafka] junrao commented on a change in pull request #11060: MINOR Refactored the existing CheckpointFile in core module, moved to server-common module and introduced it as SnapshotFile.

[GitBox <gi...@apache.org>]

junrao commented on a change in pull request #11060:
URL: https://github.com/apache/kafka/pull/11060#discussion_r671517216



##########
File path: core/src/main/scala/kafka/server/checkpoints/CheckpointFile.scala
##########
@@ -16,127 +16,41 @@
   */
 package kafka.server.checkpoints
 
-import java.io._
-import java.nio.charset.StandardCharsets
-import java.nio.file.{FileAlreadyExistsException, Files, Paths}
-
 import kafka.server.LogDirFailureChannel
-import kafka.utils.Logging
 import org.apache.kafka.common.errors.KafkaStorageException
-import org.apache.kafka.common.utils.Utils
-
-import scala.collection.{Seq, mutable}
+import org.apache.kafka.server.common.SnapshotFile
+import org.apache.kafka.server.common.SnapshotFile.EntryFormatter
 
-trait CheckpointFileFormatter[T]{
-  def toLine(entry: T): String
-
-  def fromLine(line: String): Option[T]
-}
-
-class CheckpointReadBuffer[T](location: String,
-                              reader: BufferedReader,
-                              version: Int,
-                              formatter: CheckpointFileFormatter[T]) extends Logging {
-  def read(): Seq[T] = {
-    def malformedLineException(line: String) =
-      new IOException(s"Malformed line in checkpoint file ($location): '$line'")
-
-    var line: String = null
-    try {
-      line = reader.readLine()
-      if (line == null)
-        return Seq.empty
-      line.toInt match {
-        case fileVersion if fileVersion == version =>
-          line = reader.readLine()
-          if (line == null)
-            return Seq.empty
-          val expectedSize = line.toInt
-          val entries = mutable.Buffer[T]()
-          line = reader.readLine()
-          while (line != null) {
-            val entry = formatter.fromLine(line)
-            entry match {
-              case Some(e) =>
-                entries += e
-                line = reader.readLine()
-              case _ => throw malformedLineException(line)
-            }
-          }
-          if (entries.size != expectedSize)
-            throw new IOException(s"Expected $expectedSize entries in checkpoint file ($location), but found only ${entries.size}")
-          entries
-        case _ =>
-          throw new IOException(s"Unrecognized version of the checkpoint file ($location): " + version)
-      }
-    } catch {
-      case _: NumberFormatException => throw malformedLineException(line)
-    }
-  }
-}
+import java.io._
+import scala.collection.Seq
+import scala.jdk.CollectionConverters._
 
 class CheckpointFile[T](val file: File,
                         version: Int,
-                        formatter: CheckpointFileFormatter[T],
+                        formatter: EntryFormatter[T],
                         logDirFailureChannel: LogDirFailureChannel,
-                        logDir: String) extends Logging {
-  private val path = file.toPath.toAbsolutePath
-  private val tempPath = Paths.get(path.toString + ".tmp")
-  private val lock = new Object()
-
-  try Files.createFile(file.toPath) // create the file if it doesn't exist
-  catch { case _: FileAlreadyExistsException => }
+                        logDir: String) {

Review comment:
       Do we need CheckpointFile? Could we just extend SnapshotFile to take an IOExceptionHandler and logDir?




-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org