public static Map<String, String> getStreamConfigMap(TableConfig tableConfig) {
    String tableNameWithType = tableConfig.getTableName();
    Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME,
        "Cannot fetch streamConfigs for OFFLINE table: %s", tableNameWithType);
    Map<String, String> streamConfigMap = null;
    if (tableConfig.getIngestionConfig() != null
        && tableConfig.getIngestionConfig().getStreamIngestionConfig() != null) {
      List<Map<String, String>> streamConfigMaps =
          tableConfig.getIngestionConfig().getStreamIngestionConfig().getStreamConfigMaps();
      Preconditions.checkState(streamConfigMaps.size() == 1, "Only 1 stream supported per table");
      streamConfigMap = streamConfigMaps.get(0);
    }
    if (streamConfigMap == null && tableConfig.getIndexingConfig() != null) {
      streamConfigMap = tableConfig.getIndexingConfig().getStreamConfigs();
    }
    if (streamConfigMap == null) {
      throw new IllegalStateException("Could not find streamConfigs for REALTIME table: " + tableNameWithType);
    }
    return streamConfigMap;
  }
@Test
  public void testGetStreamConfigMap() {
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
    try {
      IngestionConfigUtils.getStreamConfigMap(tableConfig);
      Assert.fail("Should fail for OFFLINE table");
    } catch (IllegalStateException e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName("myTable").setTimeColumnName("timeColumn").build();

    // get from ingestion config (when not present in indexing config)
    Map<String, String> streamConfigMap = Collections.singletonMap("streamType", "kafka");
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setStreamIngestionConfig(new StreamIngestionConfig(Collections.singletonList(streamConfigMap)));
    tableConfig.setIngestionConfig(ingestionConfig);
    Map<String, String> actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
    Assert.assertEquals(actualStreamConfigsMap.size(), 1);
    Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "kafka");

    // get from ingestion config (even if present in indexing config)
    Map<String, String> deprecatedStreamConfigMap = new HashMap<>();
    deprecatedStreamConfigMap.put("streamType", "foo");
    deprecatedStreamConfigMap.put("customProp", "foo");
    IndexingConfig indexingConfig = new IndexingConfig();
    indexingConfig.setStreamConfigs(deprecatedStreamConfigMap);
    tableConfig.setIndexingConfig(indexingConfig);
    actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
    Assert.assertEquals(actualStreamConfigsMap.size(), 1);
    Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "kafka");

    // fail if multiple found
    ingestionConfig.setStreamIngestionConfig(
        new StreamIngestionConfig(Arrays.asList(streamConfigMap, deprecatedStreamConfigMap)));
    try {
      IngestionConfigUtils.getStreamConfigMap(tableConfig);
      Assert.fail("Should fail for multiple stream configs");
    } catch (IllegalStateException e) {
      // expected
    }

    // get from indexing config
    tableConfig.setIngestionConfig(null);
    actualStreamConfigsMap = IngestionConfigUtils.getStreamConfigMap(tableConfig);
    Assert.assertEquals(actualStreamConfigsMap.size(), 2);
    Assert.assertEquals(actualStreamConfigsMap.get("streamType"), "foo");

    // fail if found nowhere
    tableConfig.setIndexingConfig(null);
    try {
      IngestionConfigUtils.getStreamConfigMap(tableConfig);
      Assert.fail("Should fail for no stream config found");
    } catch (IllegalStateException e) {
      // expected
    }
  }