public static TableUpsertMetadataManager create(TableConfig tableConfig,
      @Nullable PinotConfiguration instanceUpsertConfig) {
    String tableNameWithType = tableConfig.getTableName();
    UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
    Preconditions.checkArgument(upsertConfig != null, "Must provide upsert config for table: %s", tableNameWithType);

    TableUpsertMetadataManager metadataManager;
    String metadataManagerClass = upsertConfig.getMetadataManagerClass();

    if (instanceUpsertConfig != null) {
      if (metadataManagerClass == null) {
        metadataManagerClass = instanceUpsertConfig.getProperty(UPSERT_DEFAULT_METADATA_MANAGER_CLASS);
      }
      // Server level config honoured only when table level config is not set to true
      if (!upsertConfig.isEnableSnapshot()) {
        upsertConfig.setEnableSnapshot(
            Boolean.parseBoolean(instanceUpsertConfig.getProperty(UPSERT_DEFAULT_ENABLE_SNAPSHOT, "false")));
      }

      // Server level config honoured only when table level config is not set to true
      if (!upsertConfig.isEnablePreload()) {
        upsertConfig.setEnablePreload(
            Boolean.parseBoolean(instanceUpsertConfig.getProperty(UPSERT_DEFAULT_ENABLE_PRELOAD, "false")));
      }

      // server level config honoured only when table level config is not set to true
      if (!upsertConfig.isAllowPartialUpsertConsumptionDuringCommit()) {
        upsertConfig.setAllowPartialUpsertConsumptionDuringCommit(Boolean.parseBoolean(
            instanceUpsertConfig.getProperty(UPSERT_DEFAULT_ALLOW_PARTIAL_UPSERT_CONSUMPTION_DURING_COMMIT, "false")));
      }
    }

    if (StringUtils.isNotEmpty(metadataManagerClass)) {
      LOGGER.info("Creating TableUpsertMetadataManager with class: {} for table: {}", metadataManagerClass,
          tableNameWithType);
      try {
        metadataManager = (TableUpsertMetadataManager) Class.forName(metadataManagerClass).newInstance();
      } catch (Exception e) {
        throw new RuntimeException(
            String.format("Caught exception while constructing TableUpsertMetadataManager with class: %s for table: %s",
                metadataManagerClass, tableNameWithType), e);
      }
    } else {
      LOGGER.info("Creating ConcurrentMapTableUpsertMetadataManager for table: {}", tableNameWithType);
      metadataManager = new ConcurrentMapTableUpsertMetadataManager();
    }

    return metadataManager;
  }
@Test
  public void testCreateForDefaultManagerClass() {
    UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setHashFunction(HashFunction.NONE);
    _tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setUpsertConfig(upsertConfig).build();
    TableUpsertMetadataManager tableUpsertMetadataManager =
        TableUpsertMetadataManagerFactory.create(_tableConfig, null);
    assertNotNull(tableUpsertMetadataManager);
    assertTrue(tableUpsertMetadataManager instanceof ConcurrentMapTableUpsertMetadataManager);
  }