Fixed integration test failure and addressed comments

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/selector/impl/DefaultStorageSelector.java

@@ -4,6 +4,7 @@
 import com.linkedin.openhouse.cluster.storage.StorageManager;
 import com.linkedin.openhouse.cluster.storage.selector.BaseStorageSelector;
 import com.linkedin.openhouse.cluster.storage.selector.StorageSelector;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -12,6 +13,7 @@
  * for the cluster in yaml configuration for all tables
  */
 @Component
+@Slf4j
 public class DefaultStorageSelector extends BaseStorageSelector {
 
   @Autowired private StorageManager storageManager;
@@ -25,6 +27,8 @@ public class DefaultStorageSelector extends BaseStorageSelector {
    */
   @Override
   public Storage selectStorage(String db, String table) {
-    return storageManager.getDefaultStorage();
+    Storage storage = storageManager.getDefaultStorage();
+    log.info("Selected storage type={} for {}.{}", storage.getType().getValue(), db, table);
+    return storage;
   }
 }

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalCatalog.java

@@ -12,6 +12,7 @@
 import com.linkedin.openhouse.internal.catalog.model.HouseTable;
 import com.linkedin.openhouse.internal.catalog.model.HouseTablePrimaryKey;
 import com.linkedin.openhouse.internal.catalog.repository.HouseTableRepository;
+import com.linkedin.openhouse.internal.catalog.repository.exception.HouseTableNotFoundException;
 import com.linkedin.openhouse.internal.catalog.repository.exception.HouseTableRepositoryException;
 import com.linkedin.openhouse.internal.catalog.toggle.IcebergFeatureGate;
 import io.micrometer.core.instrument.MeterRegistry;
@@ -64,12 +65,12 @@ public class OpenHouseInternalCatalog extends BaseMetastoreCatalog {
   protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
     return new OpenHouseInternalTableOperations(
         houseTableRepository,
-        getFileIO(tableIdentifier),
+        fileIOManager,
+        resolveFileIO(tableIdentifier),
         snapshotInspector,
         houseTableMapper,
         tableIdentifier,
-        new MetricsReporter(this.meterRegistry, METRICS_PREFIX, Lists.newArrayList()),
-        fileIOManager);
+        new MetricsReporter(this.meterRegistry, METRICS_PREFIX, Lists.newArrayList()));
   }
 
   /** Overwritten for annotation purpose. */
@@ -155,13 +156,21 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
    * @param tableIdentifier
    * @return fileIO
    */
-  private FileIO getFileIO(TableIdentifier tableIdentifier) {
-    Optional<HouseTable> houseTable =
-        houseTableRepository.findById(
-            HouseTablePrimaryKey.builder()
-                .databaseId(tableIdentifier.namespace().toString())
-                .tableId(tableIdentifier.name())
-                .build());
+  private FileIO resolveFileIO(TableIdentifier tableIdentifier) {
+    Optional<HouseTable> houseTable = Optional.empty();
+    try {
+      houseTable =
+          houseTableRepository.findById(
+              HouseTablePrimaryKey.builder()
+                  .databaseId(tableIdentifier.namespace().toString())
+                  .tableId(tableIdentifier.name())
+                  .build());
+    } catch (HouseTableNotFoundException e) {
+      log.info(
+          "House table not found {}.{}",
+          tableIdentifier.namespace().toString(),
+          tableIdentifier.name());
+    }
     StorageType.Type type =
         houseTable.isPresent()
             ? storageType.fromString(houseTable.get().getStorageType())

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperations.java

@@ -51,6 +51,8 @@ public class OpenHouseInternalTableOperations extends BaseMetastoreTableOperatio
 
   HouseTableRepository houseTableRepository;
 
+  FileIOManager fileIOManager;
+
   FileIO fileIO;
 
   SnapshotInspector snapshotInspector;
@@ -61,8 +63,6 @@ public class OpenHouseInternalTableOperations extends BaseMetastoreTableOperatio
 
   MetricsReporter metricsReporter;
 
-  FileIOManager fileIOManager;
-
   private static final Gson GSON = new Gson();
 
   private static final Cache<String, Integer> CACHE =
@@ -228,7 +228,7 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
 
       houseTable = houseTableMapper.toHouseTable(updatedMetadata);
       // Set the storage type of table in house table before persisting in hts
-      houseTable.setStorageType(fileIOManager.getStorage(io()).getType().getValue());
+      houseTable.setStorageType(fileIOManager.getStorage(fileIO).getType().getValue());
       if (!isStageCreate) {
         houseTableRepository.save(houseTable);
       } else {

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperationsTest.java

@@ -84,16 +84,15 @@ void setup() {
     openHouseInternalTableOperations =
         new OpenHouseInternalTableOperations(
             mockHouseTableRepository,
+            fileIOManager,
             fileIO,
             Mockito.mock(SnapshotInspector.class),
             mockHouseTableMapper,
             TEST_TABLE_IDENTIFIER,
-            new MetricsReporter(new SimpleMeterRegistry(), "TEST_CATALOG", Lists.newArrayList()),
-            fileIOManager);
+            new MetricsReporter(new SimpleMeterRegistry(), "TEST_CATALOG", Lists.newArrayList()));
     LocalStorage localStorage = mock(LocalStorage.class);
     when(fileIOManager.getStorage(fileIO)).thenReturn(localStorage);
     when(localStorage.getType()).thenReturn(StorageType.LOCAL);
-    // when(localStorage.getType()).thenReturn(StorageType.LOCAL);
   }
 
   @Test

services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/RepositoryTestWithSettableComponents.java

@@ -100,12 +100,12 @@ void testNoRetryInternalRepo() {
     OpenHouseInternalTableOperations actualOps =
         new OpenHouseInternalTableOperations(
             houseTablesRepository,
+            fileIOManager,
             fileIO,
             snapshotInspector,
             houseTableMapper,
             tableIdentifier,
-            new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()),
-            fileIOManager);
+            new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()));
     ((SettableCatalogForTest) catalog).setOperation(actualOps);
     TableDto creationDTO = TABLE_DTO.toBuilder().tableVersion(INITIAL_TABLE_VERSION).build();
     creationDTO = openHouseInternalRepository.save(creationDTO);
@@ -117,12 +117,12 @@ void testNoRetryInternalRepo() {
     OpenHouseInternalTableOperations mockOps =
         new OpenHouseInternalTableOperations(
             htsRepo,
+            fileIOManager,
             fileIO,
             snapshotInspector,
             houseTableMapper,
             tableIdentifier,
-            new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()),
-            fileIOManager);
+            new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()));
     OpenHouseInternalTableOperations spyOperations = Mockito.spy(mockOps);
     doReturn(actualOps.current()).when(spyOperations).refresh();
     BaseTable spyOptsMockedTable = Mockito.spy(new BaseTable(spyOperations, realTable.name()));
@@ -198,12 +198,12 @@ void testFailedHtsRepoWhenGet() {
       OpenHouseInternalTableOperations mockOps =
           new OpenHouseInternalTableOperations(
               htsRepo,
+              fileIOManager,
               fileIO,
               snapshotInspector,
               houseTableMapper,
               tableIdentifier,
-              new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()),
-              fileIOManager);
+              new MetricsReporter(this.meterRegistry, "test", Lists.newArrayList()));
       OpenHouseInternalTableOperations spyOperations = Mockito.spy(mockOps);
       BaseTable spyOptsMockedTable =
           Mockito.spy(