Close metrics reporter in RESTSessionCatalog and add test in CatalogTests (#16310)

Author: talatuyarer

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -269,6 +269,7 @@ public void initialize(String name, Map<String, String> unresolved) {
                 .toUpperCase(Locale.US));
 
     this.reporter = CatalogUtil.loadMetricsReporter(mergedProps);
+    this.closeables.addCloseable(reporter);
 
     this.reportingViaRestEnabled =
         PropertyUtil.propertyAsBoolean(

core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java

@@ -24,6 +24,7 @@
 import static org.assertj.core.api.Assertions.setMaxStackTraceElementsDisplayed;
 import static org.assertj.core.api.Assumptions.assumeThat;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
@@ -3391,11 +3392,18 @@ public void testCatalogWithCustomMetricsReporter() throws IOException {
     assertThat(CustomMetricsReporter.SCAN_COUNTER.get()).isEqualTo(1);
     // reset counter in case subclasses run this test multiple times
     CustomMetricsReporter.SCAN_COUNTER.set(0);
+
+    CustomMetricsReporter.CLOSE_COUNTER.set(0);
+    ((Closeable) catalogWithCustomReporter).close();
+    assertThat(CustomMetricsReporter.CLOSE_COUNTER.get())
+        .as("Catalog.close() must propagate to the configured MetricsReporter")
+        .isEqualTo(1);
   }
 
   public static class CustomMetricsReporter implements MetricsReporter {
     static final AtomicInteger SCAN_COUNTER = new AtomicInteger(0);
     static final AtomicInteger COMMIT_COUNTER = new AtomicInteger(0);
+    static final AtomicInteger CLOSE_COUNTER = new AtomicInteger(0);
 
     @Override
     public void report(MetricsReport report) {
@@ -3405,6 +3413,11 @@ public void report(MetricsReport report) {
         COMMIT_COUNTER.incrementAndGet();
       }
     }
+
+    @Override
+    public void close() {
+      CLOSE_COUNTER.incrementAndGet();
+    }
   }
 
   private static void assertEmpty(String context, Catalog catalog, Namespace ns) {