public static FlinkJobServerDriver fromConfig(FlinkServerConfiguration configuration) {
    return create(
        configuration,
        createJobServerFactory(configuration),
        createArtifactServerFactory(configuration),
        () -> FlinkJobInvoker.create(configuration));
  }
@Test
  public void testConfigurationDefaults() {
    FlinkJobServerDriver.FlinkServerConfiguration config =
        new FlinkJobServerDriver.FlinkServerConfiguration();
    assertThat(config.getHost(), is("localhost"));
    assertThat(config.getPort(), is(8099));
    assertThat(config.getArtifactPort(), is(8098));
    assertThat(config.getExpansionPort(), is(8097));
    assertThat(config.getFlinkMaster(), is("[auto]"));
    assertThat(config.isCleanArtifactsPerJob(), is(true));
    FlinkJobServerDriver flinkJobServerDriver = FlinkJobServerDriver.fromConfig(config);
    assertThat(flinkJobServerDriver, is(not(nullValue())));
  }