public void startServices() throws ExecutionException, InterruptedException {
    LOG.info("StartServices Config: " + cfg);
    List<String> tablePaths;
    if (cfg.autoDiscovery) {
      // We support defining multi base paths
      tablePaths = cfg.basePath.stream()
          .filter(this::pathExists)
          .flatMap(p -> MultiTableServiceUtils.findHoodieTablesUnderPath(jsc, p).stream())
          .collect(Collectors.toList());
    } else {
      tablePaths = MultiTableServiceUtils.getTablesToBeServedFromProps(jsc, props);
    }
    LOG.info("All table paths: " + String.join(",", tablePaths));
    if (cfg.batch) {
      batchRunTableServices(tablePaths);
    } else {
      streamRunTableServices(tablePaths);
    }
  }
@Test
  public void testRunAllServicesForSingleTable() throws IOException, ExecutionException, InterruptedException {
    HoodieMultiTableServicesMain.Config cfg = getHoodieMultiServiceConfig();
    HoodieTableMetaClient metaClient1 = getMetaClient("table1");
    cfg.batch = true;
    cfg.basePath = Collections.singletonList(metaClient1.getBasePath().toString());
    HoodieMultiTableServicesMain main = new HoodieMultiTableServicesMain(jsc, cfg);
    main.startServices();
    // Verify cleans
    Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCleanerTimeline().countInstants());
    // Verify delta commits
    Assertions.assertEquals(2, metaClient1.reloadActiveTimeline().getDeltaCommitTimeline().countInstants());
    // Verify replace commits
    Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCompletedReplaceTimeline().countInstants());
    // Verify compactions, delta commits and replace commits
    Assertions.assertEquals(4, metaClient1.reloadActiveTimeline().getCommitsTimeline().countInstants());
  }