Respond to comments

build.sbt

@@ -658,6 +658,7 @@ lazy val kernelDefaults = (project in file("kernel/kernel-defaults"))
     libraryDependencies ++= Seq(
       "org.apache.hadoop" % "hadoop-client-runtime" % hadoopVersion,
       "com.fasterxml.jackson.core" % "jackson-databind" % "2.13.5",
+      "com.fasterxml.jackson.datatype" % "jackson-datatype-jdk8" % "2.13.5",
       "org.apache.parquet" % "parquet-hadoop" % "1.12.3",
 
       "org.scalatest" %% "scalatest" % scalaTestVersion % "test",

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/metrics/MetricsReportSerializerSuite.scala

@@ -66,9 +66,9 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
     val snapshotReport1 = new SnapshotReportImpl(
       "/table/path",
       snapshotMetrics1,
-      Optional.of(1),
-      Optional.of(0),
-      Optional.of(exception)
+      Optional.of(1), // version
+      Optional.of(0), // providedTimestamp
+      Optional.of(exception) // exception
     )
 
     // Manually check expected JSON
@@ -96,9 +96,9 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
     val snapshotReport2 = new SnapshotReportImpl(
       "/table/path",
       snapshotMetrics2,
-      Optional.empty(),
-      Optional.empty(),
-      Optional.empty()
+      Optional.empty(), // version
+      Optional.empty(), // providedTimestamp
+      Optional.empty() // exception
     )
     testSnapshotReport(snapshotReport2)
   }

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/engine/LoggingMetricsReporter.java

@@ -18,7 +18,6 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import io.delta.kernel.engine.MetricsReporter;
 import io.delta.kernel.internal.metrics.MetricsReportSerializers;
-import io.delta.kernel.internal.snapshot.SnapshotManager;
 import io.delta.kernel.metrics.MetricsReport;
 import io.delta.kernel.metrics.SnapshotReport;
 import org.slf4j.Logger;
@@ -30,22 +29,23 @@
  */
 public class LoggingMetricsReporter implements MetricsReporter {
 
-  private static final Logger logger = LoggerFactory.getLogger(SnapshotManager.class);
+  private static final Logger logger = LoggerFactory.getLogger(LoggingMetricsReporter.class);
 
   @Override
   public void report(MetricsReport report) {
     try {
       if (report instanceof SnapshotReport) {
         logger.info(
-            "SnapshotReport = %s",
+            "SnapshotReport = {}",
             MetricsReportSerializers.serializeSnapshotReport((SnapshotReport) report));
       } else {
         logger.info(
-            "%s = [%s does not support serializing this type of MetricReport]",
-            report.getClass(), this.getClass());
+            "{} = [{} does not support serializing this type of MetricReport]",
+            report.getClass(),
+            this.getClass());
       }
     } catch (JsonProcessingException e) {
-      logger.info("Encountered exception while serializing report %s: %s", report, e);
+      logger.info("Encountered exception while serializing report {}: {}", report, e);
     }
   }
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/MetricsReportSuite.scala

@@ -360,6 +360,11 @@ class MetricsReportSuite extends AnyFunSuite with TestUtils {
       override def report(report: MetricsReport): Unit = buf.append(report)
     }
 
+    private val metricsReporters = new util.ArrayList[MetricsReporter]() {{
+      addAll(baseEngine.getMetricsReporters)
+      add(metricsReporter)
+    }}
+
     override def getExpressionHandler: ExpressionHandler = baseEngine.getExpressionHandler
 
     override def getJsonHandler: JsonHandler = baseEngine.getJsonHandler
@@ -373,7 +378,7 @@ class MetricsReportSuite extends AnyFunSuite with TestUtils {
       baseEngine.getCommitCoordinatorClientHandler(name, conf)
 
     override def getMetricsReporters(): java.util.List[MetricsReporter] = {
-      java.util.Collections.singletonList(metricsReporter)
+      metricsReporters
     }
   }
 }