public static StandaloneExecutor create(
      final Map<String, String> properties,
      final String queriesFile,
      final String installDir,
      final MetricCollectors metricCollectors
  ) {
    final KsqlConfig tempConfig = new KsqlConfig(properties);
    
    final Function<KsqlConfig, ServiceContext> serviceContextFactory = 
        config -> ServiceContextFactory.create(config, DisabledKsqlClient::instance);
    final ServiceContext tempServiceContext =
        serviceContextFactory.apply(tempConfig);

    final String kafkaClusterId = KafkaClusterUtil.getKafkaClusterId(tempServiceContext);
    final String ksqlServerId = tempConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG);
    final Map<String, Object> updatedProperties = tempConfig.originals();
    updatedProperties.putAll(
        metricCollectors.addConfluentMetricsContextConfigs(ksqlServerId, kafkaClusterId));

    final Consumer<KsqlConfig> rocksDBConfigSetterHandler =
        RocksDBConfigSetterHandler::maybeConfigureRocksDBConfigSetter;

    return create(
        updatedProperties,
        queriesFile,
        installDir,
        serviceContextFactory,
        KafkaConfigStore::new,
        KsqlVersionCheckerAgent::new,
        StandaloneExecutor::new,
        metricCollectors,
        rocksDBConfigSetterHandler
    );
  }
@SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT") // Not a bug
  @Test
  public void shouldCreateConfigTopicThenGetConfig() {
    // When:
    create();

    // Then:
    final InOrder inOrder = Mockito.inOrder(topicClient, configStoreFactory, constructor);
    inOrder.verify(topicClient).createTopic(eq(configTopicName), anyInt(), anyShort(), anyMap());
    inOrder.verify(configStoreFactory).apply(eq(configTopicName), argThat(sameConfig(baseConfig)));
    inOrder.verify(constructor).create(
        any(), any(), same(mergedConfig), any(), anyString(), any(), anyBoolean(), any(), any(), any(), any());

    argumentCaptor.getValue().close();
  }