class KsqlRestApplication extends Application<KsqlRestConfig> implements Executable {

  static void ensureCommandTopic(final KsqlRestConfig restConfig,
                                 final KafkaTopicClient topicClient,
                                 final String commandTopic) {
    final long requiredTopicRetention = Long.MAX_VALUE;
    if (topicClient.isTopicExists(commandTopic)) {
      final ImmutableMap<String, Object> requiredConfig =
          ImmutableMap.of(TopicConfig.RETENTION_MS_CONFIG, requiredTopicRetention);

      if (topicClient.addTopicConfig(commandTopic, requiredConfig)) {
        log.info("Corrected retention.ms on command topic. topic:{}, retention.ms:{}",
                 commandTopic, requiredTopicRetention);
      }

      return;
    }

    try {
      short replicationFactor = 1;
      if (restConfig.getOriginals().containsKey(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY)) {
        replicationFactor = Short.parseShort(
            restConfig
                .getOriginals()
                .get(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY)
                .toString()
        );
      }
      if (replicationFactor < 2) {
        log.warn("Creating topic {} with replication factor of %d which is less than 2. "
                 + "This is not advisable in a production environment. ", replicationFactor);
      }

      // for now we create the command topic with infinite retention so that we
      // don't lose any data in case of fail over etc.
      topicClient.createTopic(
          commandTopic,
          1,
          replicationFactor,
          Collections.singletonMap(TopicConfig.RETENTION_MS_CONFIG, requiredTopicRetention)
      );
    } catch (KafkaTopicException e) {
      log.info("Command Topic Exists: {}", e.getMessage());
    }
  }

  private  KsqlRestApplication(
      KsqlEngine ksqlEngine,
      KsqlRestConfig config,
      CommandRunner commandRunner,
      RootDocument rootDocument,
      StatusResource statusResource,
      StreamedQueryResource streamedQueryResource,
      KsqlResource ksqlResource,
      boolean isUiEnabled,
      VersionCheckerAgent versionCheckerAgent,
      ServerInfo serverInfo
  );

}