public static void convertFromLegacyTableConfig(TableConfig tableConfig) {
    // It is possible that indexing as well as ingestion configs exist, in which case we always honor ingestion config.
    IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
    BatchIngestionConfig batchIngestionConfig =
        (ingestionConfig != null) ? ingestionConfig.getBatchIngestionConfig() : null;

    SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
    String segmentPushType = validationConfig.getSegmentPushType();
    String segmentPushFrequency = validationConfig.getSegmentPushFrequency();

    if (batchIngestionConfig == null) {
      // Only create the config if any of the deprecated config is not null.
      if (segmentPushType != null || segmentPushFrequency != null) {
        batchIngestionConfig = new BatchIngestionConfig(null, segmentPushType, segmentPushFrequency);
      }
    } else {
      // This should not happen typically, but since we are in repair mode, might as well cover this corner case.
      if (batchIngestionConfig.getSegmentIngestionType() == null) {
        batchIngestionConfig.setSegmentIngestionType(segmentPushType);
      }
      if (batchIngestionConfig.getSegmentIngestionFrequency() == null) {
        batchIngestionConfig.setSegmentIngestionFrequency(segmentPushFrequency);
      }
    }

    StreamIngestionConfig streamIngestionConfig =
        (ingestionConfig != null) ? ingestionConfig.getStreamIngestionConfig() : null;
    IndexingConfig indexingConfig = tableConfig.getIndexingConfig();

    if (streamIngestionConfig == null) {
      // Only set the new config if the deprecated one is set.
      Map<String, String> streamConfigs = indexingConfig.getStreamConfigs();
      if (MapUtils.isNotEmpty(streamConfigs)) {
        streamIngestionConfig = new StreamIngestionConfig(Collections.singletonList(streamConfigs));
      }
    }

    if (ingestionConfig == null) {
      if (batchIngestionConfig != null || streamIngestionConfig != null) {
        ingestionConfig = new IngestionConfig();
        ingestionConfig.setBatchIngestionConfig(batchIngestionConfig);
        ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
      }
    } else {
      ingestionConfig.setBatchIngestionConfig(batchIngestionConfig);
      ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
    }

    // Set the new config fields.
    tableConfig.setIngestionConfig(ingestionConfig);

    // Clear the deprecated ones.
    indexingConfig.setStreamConfigs(null);
    validationConfig.setSegmentPushFrequency(null);
    validationConfig.setSegmentPushType(null);
  }
@Test
  public void testConvertFromLegacyTableConfig() {
    String expectedPushFrequency = "HOURLY";
    String expectedPushType = "APPEND";

    Map<String, String> expectedStreamConfigsMap = getTestStreamConfigs();
    TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME)
        .setSegmentPushFrequency(expectedPushFrequency).setSegmentPushType(expectedPushType)
        .setStreamConfigs(expectedStreamConfigsMap).build();

    // Before conversion, the ingestion config should be null.
    Assert.assertNull(tableConfig.getIngestionConfig());

    // Perform conversion.
    TableConfigUtils.convertFromLegacyTableConfig(tableConfig);

    // After conversion, assert that the configs are transferred ingestionConfig.
    BatchIngestionConfig batchIngestionConfig = tableConfig.getIngestionConfig().getBatchIngestionConfig();
    Assert.assertEquals(batchIngestionConfig.getSegmentIngestionFrequency(), expectedPushFrequency);
    Assert.assertEquals(batchIngestionConfig.getSegmentIngestionType(), expectedPushType);

    Map<String, String> actualStreamConfigsMap =
        tableConfig.getIngestionConfig().getStreamIngestionConfig().getStreamConfigMaps().get(0);
    Assert.assertEquals(actualStreamConfigsMap, expectedStreamConfigsMap);

    // Assert that the deprecated fields are cleared.
    Assert.assertNull(tableConfig.getIndexingConfig().getStreamConfigs());

    SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
    Assert.assertNull(validationConfig.getSegmentPushFrequency());
    Assert.assertNull(validationConfig.getSegmentPushType());
  }