Fix NPE in CreateCatalog (#2435)

Author: HonahX

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -781,6 +781,13 @@ public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
       throw new AlreadyExistsException(
           "Cannot create Catalog %s. Catalog already exists or resolution failed",
           entity.getName());
+    } else if (!catalogResult.isSuccess()) {
+      throw new IllegalStateException(
+          String.format(
+              "Cannot create Catalog %s: %s with extraInfo %s",
+              entity.getName(),
+              catalogResult.getReturnStatus(),
+              catalogResult.getExtraInformation()));
     }
     return PolarisEntity.of(catalogResult.getCatalog());
   }

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java

@@ -164,7 +164,7 @@ public Response createCatalog(
     Catalog catalog = request.getCatalog();
     validateStorageConfig(catalog.getStorageConfigInfo());
     validateExternalCatalog(catalog);
-    Catalog newCatalog = new CatalogEntity(adminService.createCatalog(request)).asCatalog();
+    Catalog newCatalog = CatalogEntity.of(adminService.createCatalog(request)).asCatalog();
     LOGGER.info("Created new catalog {}", newCatalog);
     return Response.status(Response.Status.CREATED).build();
   }

runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java

@@ -49,13 +49,16 @@
 import org.apache.polaris.core.entity.PrincipalRoleEntity;
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.dao.entity.BaseResult;
 import org.apache.polaris.core.persistence.dao.entity.CreateCatalogResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.secrets.UnsafeInMemorySecretsManager;
 import org.apache.polaris.service.TestServices;
 import org.apache.polaris.service.config.ReservedProperties;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 public class ManagementServiceTest {
   private TestServices services;
@@ -285,4 +288,44 @@ public void testCanListCatalogs() {
         .extracting(Catalog::getName)
         .containsExactlyInAnyOrder("my-catalog-1", "my-catalog-2");
   }
+
+  @Test
+  public void testCreateCatalogReturnErrorOnFailure() {
+    PolarisMetaStoreManager metaStoreManager = Mockito.spy(setupMetaStoreManager());
+    PolarisCallContext callContext = services.newCallContext();
+    PolarisAdminService polarisAdminService =
+        setupPolarisAdminService(metaStoreManager, callContext);
+
+    AwsStorageConfigInfo awsConfigModel =
+        AwsStorageConfigInfo.builder()
+            .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+            .setExternalId("externalId")
+            .setUserArn("userArn")
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+            .setAllowedLocations(List.of("s3://my-old-bucket/path/to/data"))
+            .build();
+    String catalogName = "mycatalog";
+    Catalog catalog =
+        PolarisCatalog.builder()
+            .setType(Catalog.TypeEnum.INTERNAL)
+            .setName(catalogName)
+            .setProperties(new CatalogProperties("s3://bucket/path/to/data"))
+            .setStorageConfigInfo(awsConfigModel)
+            .build();
+    CreateCatalogResult resultWithError =
+        new CreateCatalogResult(
+            BaseResult.ReturnStatus.UNEXPECTED_ERROR_SIGNALED, "Unexpected Error Occurred");
+    Mockito.doAnswer(invocation -> resultWithError)
+        .when(metaStoreManager)
+        .createCatalog(Mockito.any(), Mockito.any(), Mockito.any());
+    Assertions.assertThatThrownBy(
+            () -> polarisAdminService.createCatalog(new CreateCatalogRequest(catalog)))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessage(
+            String.format(
+                "Cannot create Catalog %s: %s with extraInfo %s",
+                catalogName,
+                resultWithError.getReturnStatus(),
+                resultWithError.getExtraInformation()));
+  }
 }