@Override
  public Catalog createCatalog(Context context) {
    final FactoryUtil.CatalogFactoryHelper helper =
        FactoryUtil.createCatalogFactoryHelper(this, context);
    helper.validateExcept(HadoopConfigurations.HADOOP_PREFIX);
    String mode = helper.getOptions().get(CatalogOptions.MODE);
    switch (mode.toLowerCase(Locale.ROOT)) {
      case "hms":
        return new HoodieHiveCatalog(
            context.getName(),
            (Configuration) helper.getOptions());
      case "dfs":
        return new HoodieCatalog(
            context.getName(),
            (Configuration) helper.getOptions());
      default:
        throw new HoodieCatalogException(String.format("Invalid catalog mode: %s, supported modes: [hms, dfs].", mode));
    }
  }
@Test
  void testCreateHMSCatalog() {
    final String catalogName = "mycatalog";

    final HoodieHiveCatalog expectedCatalog = HoodieCatalogTestUtils.createHiveCatalog(catalogName);

    final Map<String, String> options = new HashMap<>();
    options.put(CommonCatalogOptions.CATALOG_TYPE.key(), HoodieCatalogFactory.IDENTIFIER);
    options.put(CatalogOptions.HIVE_CONF_DIR.key(), CONF_DIR.getPath());
    options.put(CatalogOptions.MODE.key(), "hms");
    options.put(CatalogOptions.TABLE_EXTERNAL.key(), "false");

    final Catalog actualCatalog =
        FactoryUtil.createCatalog(
            catalogName, options, null, Thread.currentThread().getContextClassLoader());

    assertEquals(
        ((HoodieHiveCatalog) actualCatalog)
            .getHiveConf()
            .getVar(HiveConf.ConfVars.METASTOREURIS), "dummy-hms");
    checkEquals(expectedCatalog, (HoodieHiveCatalog) actualCatalog);
  }