@Override
  public void configure(final KsqlConfig config) {
    if (!config.getKsqlStreamConfigProps().containsKey(StreamsConfig.APPLICATION_SERVER_CONFIG)) {
      throw new IllegalArgumentException("Need KS application server set");
    }

    final String applicationServer =
        (String) config.getKsqlStreamConfigProps().get(StreamsConfig.APPLICATION_SERVER_CONFIG);
    final HostInfo hostInfo = ServerUtil.parseHostInfo(applicationServer);
    this.localHost = new KsqlHostInfo(hostInfo.host(), hostInfo.port());
    try {
      this.localUrl = new URL(applicationServer);
    } catch (final Exception e) {
      throw new IllegalStateException("Failed to convert remote host info to URL."
          + " remoteInfo: " + localHost.host() + ":"
          + localHost.host());
    }

    this.validator = new RequestValidator(
        CustomValidators.VALIDATOR_MAP,
        injectorFactory,
        ksqlEngine::createSandbox,
        new ValidatedCommandFactory()
    );

    this.handler = new RequestHandler(
        CustomExecutors.EXECUTOR_MAP,
        new DistributingExecutor(
            config,
            commandRunner.getCommandQueue(),
            distributedCmdResponseTimeout,
            injectorFactory,
            authorizationValidator,
            new ValidatedCommandFactory(),
            errorHandler,
            commandRunnerWarning
        ),
        ksqlEngine,
        new DefaultCommandQueueSync(
            commandRunner.getCommandQueue(),
            KsqlResource::shouldSynchronize,
            distributedCmdResponseTimeout
        )
    );
  }
@Test(expected = IllegalArgumentException.class)
  public void shouldThrowOnConfigureIfAppServerNotSet() {
    // Given:
    final KsqlConfig configNoAppServer = new KsqlConfig(ImmutableMap.of());

    // When:
    ksqlResource.configure(configNoAppServer);
  }