@Override
public Collection<T> loadConfig() {
    Collection<T> configData = new ArrayList<>();
    final File configDirectory = new File(configuration.getDataDirectory(),dataDirectory);
    final File[] dataFiles = configDirectory.listFiles((FilenameFilter) new WildcardFileFilter("*.yml"));
    if (dataFiles == null) {
        throw ConfigValidationException.createFileReadError(configDirectory.getAbsolutePath());
    }

    for (File dataFile : dataFiles) {
        T data;
        try {
            data = configurationFactory.build(dataFile);
        } catch (IOException | ConfigurationException e) {
            throw Throwables.propagate(e);
        }
        configData.add(data);

    }
    return configData;
}

@Test
public void nodeClientShouldBeCreatedFromConfig() throws URISyntaxException,IOException,ConfigurationException {
    URL configFileUrl = this.getClass().getResource("/node_client.yml");
    File configFile = new File(configFileUrl.toURI());
    EsConfiguration config = configFactory.build(configFile);

    ManagedEsClient managedEsClient = new ManagedEsClient(config);
    Client client = managedEsClient.getClient();

    assertNotNull(client);
    assertTrue(client instanceof NodeClient);

    NodeClient nodeClient = (NodeClient) client;
    assertEquals(config.getClusterName(),nodeClient.settings().get("cluster.name"));
    assertEquals("true",nodeClient.settings().get("node.client"));
    assertEquals("false",nodeClient.settings().get("node.data"));
}

@Test
public void transportClientShouldBeCreatedFromConfig() throws URISyntaxException,ConfigurationException {
    URL configFileUrl = this.getClass().getResource("/transport_client.yml");
    File configFile = new File(configFileUrl.toURI());
    EsConfiguration config = configFactory.build(configFile);

    ManagedEsClient managedEsClient = new ManagedEsClient(config);
    Client client = managedEsClient.getClient();

    assertNotNull(client);
    assertTrue(client instanceof TransportClient);

    final TransportClient transportClient = (TransportClient) client;
    assertEquals(3,transportClient.transportAddresses().size());
    assertEquals(
            TransportAddressHelper.fromHostAndPort(HostAndPort.fromParts("127.0.0.1",9300)),transportClient.transportAddresses().get(0));
    assertEquals(
            TransportAddressHelper.fromHostAndPort(HostAndPort.fromParts("127.0.0.1",9301)),transportClient.transportAddresses().get(1));
    assertEquals(
            TransportAddressHelper.fromHostAndPort(HostAndPort.fromParts("127.0.0.1",9302)),transportClient.transportAddresses().get(2));
}

@Test
public void managedClientShouldUseCustomElasticsearchConfig() throws URISyntaxException,ConfigurationException {
    URL configFileUrl = this.getClass().getResource("/custom_settings_file.yml");
    File configFile = new File(configFileUrl.toURI());
    EsConfiguration config = configFactory.build(configFile);

    ManagedEsClient managedEsClient = new ManagedEsClient(config);
    Client client = managedEsClient.getClient();

    assertNotNull(client);
    assertTrue(client instanceof NodeClient);

    NodeClient nodeClient = (NodeClient) client;
    assertEquals(config.getClusterName(),nodeClient.settings().get("cluster.name"));
    assertEquals("19300-19400",nodeClient.settings().get("transport.tcp.port"));
}

@Test
public void managedClientObeysPrecedenceOfSettings() throws URISyntaxException,ConfigurationException {
    URL configFileUrl = this.getClass().getResource("/custom_settings_precedence.yml");
    File configFile = new File(configFileUrl.toURI());
    EsConfiguration config = configFactory.build(configFile);

    ManagedEsClient managedEsClient = new ManagedEsClient(config);
    Client client = managedEsClient.getClient();

    assertNotNull(client);
    assertTrue(client instanceof NodeClient);

    NodeClient nodeClient = (NodeClient) client;
    assertEquals(config.getClusterName(),nodeClient.settings().get("cluster.name"));
    assertEquals("29300-29400",nodeClient.settings().get("transport.tcp.port"));
    assertEquals("target/data/yaml",nodeClient.settings().get("path.home"));
}

private void connect() throws URISyntaxException,ConfigurationException {
    final File yml = new File(Resources.getResource(serializer).toURI());
    final YamlConfigurationFactory<TinkerPopFactory> factory =
            new YamlConfigurationFactory<>(TinkerPopFactory.class,validator,objectMapper,"dw");
    final Cluster cluster = factory.build(yml).build();
    final Graph graph = EmptyGraph.instance();
    client = cluster.connect();
    g = graph.traversal().withRemote(DriverRemoteConnection.using(cluster,"g"));
}

@Test
public void testReadWriteClient() throws ConfigurationException,URISyntaxException {
    final Long expected = client.submit("g.addV('Person').id()").one().getLong();
    final Long person = client.submit("g.V().hasLabel('Person').id()").one().getLong();
    assertEquals(expected,person);

    if(testBytecode) {
        client.submit("g.V().drop()");
        final Long remoteExpected = (Long) g.addV("Person").id().next();
        final Long remotePerson = (Long) g.V().hasLabel("Person").id().next();
        assertEquals(remoteExpected,remotePerson);
    }
}

public SupportedMsaVersions loadSupportedMsaVersions(final URL url) {
    final ConfigurationFactory<SupportedMsaVersions> factory = supportedMsaVersionsFactoryFactory.create(
            SupportedMsaVersions.class,buildDefaultValidatorFactory().getValidator(),"");
    try {
        SupportedMsaVersions supportedMsaVersions = factory.build(
                configurationSourceProvider,url.toString());
        return supportedMsaVersions;
    } catch (IOException | ConfigurationException e) {
        throw propagate(e);
    }
}

@Test
public void canDeserializeCorrectly() throws IOException,ConfigurationException {
    final HikariDataSourceFactory factory = new YamlConfigurationFactory<>(HikariDataSourceFactory.class,BaseValidator.newValidator(),Jackson.newObjectMapper(),"dw")
            .build(new ResourceConfigurationSourceProvider(),"config.yaml");

    assertthat(factory.getUser()).isEqualTo("nick");
    assertthat(factory.getpassword()).isEqualTo("nickss");
    assertthat(factory.getDatasourceClassName()).isEqualTo("org.postgresql.ds.PGSimpleDataSource");
    assertthat(factory.getProperties()).containsExactly(entry("databaseName","postgres"));
    assertthat(factory.getMinSize()).isEqualTo(OptionalInt.empty());
    assertthat(factory.getMaxSize()).isEqualTo(12);
    assertthat(factory.isAutoCommit()).isTrue();
}

private EmoConfiguration loadConfigFile(File configFile)
        throws IOException,ConfigurationException {
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());
    ConfigurationFactory<EmoConfiguration> configurationFactory = new ConfigurationFactory(EmoConfiguration.class,mapper,"dw");
    return configurationFactory.build(configFile);
}

public static DdlConfiguration parseDdlConfiguration(File file) throws IOException,ConfigurationException {
    // Similar to Dropwizard's ConfigurationFactory but ignores System property overrides.
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());
    mapper.configure(DeserializationFeature.FAIL_ON_UNKNowN_PROPERTIES,true);
    DdlConfiguration ddlConfiguration = mapper.readValue(file,DdlConfiguration.class);

    Set<ConstraintViolation<DdlConfiguration>> errors = _validator.validate(ddlConfiguration);
    if (!errors.isEmpty()) {
        throw new ConfigurationValidationException(file.toString(),errors);
    }

    return ddlConfiguration;
}

@Test
public void ensureSdkDefaultConfigDeserialization()
        throws IOException,URISyntaxException,ConfigurationException {
    // This test makes sure that we haven't forgotten to update the emodb sdk default config file when we add/remove properties
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());
    ConfigurationFactory configurationFactory = new ConfigurationFactory(EmoConfiguration.class,"dw");
    // Make sure that our config files are up to date
    configurationFactory.build(
    new File(EmoStartMojo.class.getResource("/emodb-default-config.yaml").toURI()));
}

/**
 * Parses the given configuration file and returns a configuration object.
 *
 * @param configurationFileName The name of the configuration file.
 * @return A configuration object.
 * @throws IOException The IO error that contains detail information.
 * @throws ConfigurationException The configuration error that contains detail information.
 */
public static ApiConfiguration parse(String configurationFileName)
  throws IOException,ConfigurationException {
  if (StringUtils.isBlank(configurationFileName)) {
    throw new IllegalArgumentException("Configuration file cannot be blank");
  }

  ObjectMapper objectMapper = null;
  if (configurationFileName.endsWith("yml") || configurationFileName.endsWith("yaml")) {
    objectMapper = Jackson.newObjectMapper(new YAMLFactory());
  } else if (configurationFileName.endsWith("json")) {
    objectMapper = Jackson.newObjectMapper(new JsonFactory());
  } else {
    throw new IllegalArgumentException("Unrecognized configuration file type");
  }

  ValidatorFactory validatorFactory = Validation
      .byProvider(HibernateValidator.class)
      .configure()
      .addValidatedValueHandler(new OptionalValidatedValueUnwrapper())
      .buildValidatorFactory();

  final ConfigurationFactory<ApiConfiguration> configurationFactory =
      new DefaultConfigurationFactoryFactory<ApiConfiguration>()
        .create(ApiConfiguration.class,validatorFactory.getValidator(),"dw");

  final File file = new File(configurationFileName);
  if (!file.exists()) {
    throw new FileNotFoundException("Configuration file " + configurationFileName + " not found");
  }

  return configurationFactory.build(file);
}

static Injector createInjector() {
  KeywhizService service = new KeywhizService();
  Bootstrap<KeywhizConfig> bootstrap = new Bootstrap<>(service);
  service.initialize(bootstrap);

  File yamlFile = new File(Resources.getResource("keywhiz-test.yaml").getFile());
  Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
  ObjectMapper objectMapper = bootstrap.getobjectMapper().copy();
  KeywhizConfig config;
  try {
    config = new ConfigurationFactory<>(KeywhizConfig.class,"dw")
        .build(yamlFile);
  } catch (IOException | ConfigurationException e) {
    throw Throwables.propagate(e);
  }

  Environment environment = new Environment(service.getName(),bootstrap.getMetricRegistry(),bootstrap.getClassLoader());

  Injector injector = Guice.createInjector(new ServiceModule(config,environment));

  service.setInjector(injector);
  return injector;
}

public static BackupConfiguration loadConfiguration() throws IOException,ConfigurationException {
    if (!CONfigURATION_FILE.isPresent()) {
        throw new FileNotFoundException(String.format("Unable to find %s",CONfigURATION_FILENAME));
    }

    return CONfigURATION_FACTORY.build(CONfigURATION_FILE.get());
}

@Test
public void testTemplateConfigurationIsValid() throws IOException,ConfigurationException {
    final BackupConfiguration config = ConfigurationTestUtil.loadConfiguration();
    assertNotNull(config);

    final Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    assertTrue(validator.validate(config).isEmpty());
}

@Test
public void hasValidDefaults() throws IOException,ConfigurationException {
  final RavenAppenderFactory factory = new RavenAppenderFactory();

  assertNull("default dsn is unset",factory.getDsn());
  assertFalse("default environment is empty",factory.getEnvironment().isPresent());
  assertFalse("default extraTags is empty",factory.getEnvironment().isPresent());
  assertFalse("default ravenFactory is empty",factory.getRavenFactory().isPresent());
  assertFalse("default release is empty",factory.getRelease().isPresent());
  assertFalse("default serverName is empty",factory.getServerName().isPresent());
  assertFalse("default tags are empty",factory.getTags().isPresent());
}

@Test
public void can_set_variable_with_default_in_array() throws IOException,ConfigurationException {
    environmentProvider.put("ARRAY_2","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.array.get(2),is(equalTo("5")));
}

@Test
public void can_set_variable_with_default_in_object() throws IOException,ConfigurationException {
    environmentProvider.put("OBJECT_C","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.object.get("c"),is(equalTo("5")));
}

@Test
public void can_set_variable_without_default_in_array() throws IOException,ConfigurationException {
    environmentProvider.put("ARRAY_1","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.array.get(1),is(equalTo("5")));
}

@Test
public void can_set_variable_without_default_in_object() throws IOException,ConfigurationException {
    environmentProvider.put("OBJECT_B","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.object.get("b"),is(equalTo("5")));
}

@Test
public void uses_default_value_in_array() throws IOException,ConfigurationException {
    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.array.get(2),is(equalTo("default")));
}

@Test
public void uses_default_value_in_object() throws IOException,ConfigurationException {
    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.object.get("c"),is(equalTo("default")));
}

@Test
public void can_set_variable_with_default_in_array() throws IOException,ConfigurationException {
    environmentProvider.put("SUB_ARRAY_2","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.array.get(2),is(equalTo("5")));
}

@Test
public void can_set_variable_with_default_in_object() throws IOException,ConfigurationException {
    environmentProvider.put("SUB_OBJECT_C","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.object.get("c"),is(equalTo("5")));
}

@Test
public void can_set_variable_without_default_in_array() throws IOException,ConfigurationException {
    environmentProvider.put("SUB_ARRAY_1","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.array.get(1),is(equalTo("5")));
}

@Test
public void can_set_variable_without_default_in_object() throws IOException,ConfigurationException {
    environmentProvider.put("SUB_OBJECT_B","5");

    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.object.get("b"),is(equalTo("5")));
}

@Test
public void uses_default_value_in_array() throws IOException,ConfigurationException {
    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.array.get(2),is(equalTo("default")));
}

@Test
public void uses_default_value_in_object() throws IOException,ConfigurationException {
    TestConfiguration testConfiguration = environmentConfigurationFactory.build(new File(
            "src/test/resources/config.yaml"));

    assertthat(testConfiguration.subTestConfiguration.object.get("c"),is(equalTo("default")));
}

@Before
public void setUp() throws URISyntaxException,ConfigurationException {
    connect();
}

public String getUrl() throws ConfigurationException {
    return url;
}

@Singleton
@Provides
@Named("openDataUrl")
public String provideOpenDataClient(final OpenDataClientConfiguration configuration) throws ConfigurationException {
    return configuration.getUrl();
}

@Override
protected AzureTableMetadataStorage<BackupMetadata> getMetadataStorage() throws IOException,ConfigurationException {
    return new AzureTableMetadataStorage<>(BackupMetadata.class,configuration,azureNamespace,new MetricRegistry());
}

@Override
protected AzureTablelikeMetadataStorage<BackupMetadata> getMetadataStorage() throws IOException,ConfigurationException {
    return new AzureTablelikeMetadataStorage<>(BackupMetadata.class);
}

@Before
public void setUp() throws InvalidKeySpecException,NoSuchAlgorithmException,ConfigurationException {
    final TokenConfiguration config = ConfigurationTestUtil.loadConfiguration().getTokenConfiguration();
    generator = new TemporaryTokenGenerator(config);
}

@Override
protected MetadataStorage<BackupMetadata> getMetadataStorage() throws IOException,ConfigurationException {
    final BackupConfiguration configuration = ConfigurationTestUtil.loadConfiguration();
    return new AzureTableMetadataStorage<>(BackupMetadata.class,configuration.getoffsiteConfiguration().getStorageConfiguration(),new MetricRegistry());
}

@Override
protected AzureFileStorage getFileStorage() throws IOException,ConfigurationException,StorageException,InvalidKeyException,URISyntaxException {
    final BackupConfiguration configuration = ConfigurationTestUtil.loadConfiguration();
    return new AzureFileStorage(configuration.getoffsiteConfiguration().getStorageConfiguration(),azureNamespace);
}

@Test
public void defaultConfigShouldBeValid() throws IOException,ConfigurationException {
    configFactory.build();
}

@Test(expected = ConfigurationException.class)
public void eitherNodeClientOrServerListMustBeSet() throws IOException,URISyntaxException {
    URL configFileUrl = this.getClass().getResource("/invalid.yml");
    File configFile = new File(configFileUrl.toURI());
    configFactory.build(configFile);
}

/**
 * Loads,parses,binds,and validates a configuration object.
 *
 * @param provider the provider to to use for reading configuration files
 * @param path     the path of the configuration file
 * @return a validated configuration object
 * @throws IOException            if there is an error reading the file
 * @throws ConfigurationException if there is an error parsing or validating the file
 */
public T build(ConfigurationSourceProvider provider,String path)
    throws IOException,ConfigurationException
{
    return build(loadConfiguration(provider,path),path);
}

@Override
public RestlerConfig getConfiguration()  {
    ObjectMapper objectMapper = Jackson.newObjectMapper();

    ValidatorFactory validatorFactory = Validation
            .byProvider(HibernateValidator.class)
            .configure()
            .addValidatedValueHandler(new OptionalValidatedValueUnwrapper())
            .buildValidatorFactory();


    final ConfigurationFactory<RestlerConfig> configurationFactory =
            new DefaultConfigurationFactoryFactory<RestlerConfig>().create(RestlerConfig.class,validatorFactory.getValidator(),objectMapper,"dw");

    try {
        return configurationFactory.build(new FileConfigurationSourceProvider(),TEST_CONfig_FILE);
    } catch (Exception e) {
        throw new RuntimeException("Cannot get test configuration",e);
    }
}

@Before
public void beforeAll() throws Exception {
    server = new TestingServer();
    server.start();
    Capabilities mockCapabilities = Mockito.mock(Capabilities.class);
    when(mockCapabilities.supportsNamedVips()).thenReturn(true);
    taskFactory = new CassandraDaemonTask.Factory(mockCapabilities);
    configurationFactory = new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,BaseValidator.newValidator(),Jackson.newObjectMapper()
                            .registerModule(new GuavaModule())
                            .registerModule(new Jdk8Module()),"dw");
    connectString = server.getConnectString();
}

@Override public Statement apply(final Statement base,Description description) {
  return new Statement() {
    @Override public void evaluate() throws Throwable {
      File yamlFile = new File(Resources.getResource("keywhiz-test.yaml").getFile());
      Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
      ObjectMapper objectMapper = KeywhizService.customizeObjectMapper(Jackson.newObjectMapper());
      KeywhizConfig config = new ConfigurationFactory<>(KeywhizConfig.class,validator,"dw")
          .build(yamlFile);

      DataSource dataSource = config.getDataSourceFactory()
          .build(new MetricRegistry(),"db-migrations");

      Flyway flyway = new Flyway();
      flyway.setDataSource(dataSource);
      flyway.setLocations(config.getMigrationsDir());
      flyway.clean();
      flyway.migrate();

      DSLContext dslContext = DSLContexts.databaseAgnostic(dataSource);
      DbSeedCommand.doImport(dslContext);

      base.evaluate();
    }
  };
}

@Override
public void initialize(Bootstrap<?> bootstrap)
{
    final InjectableValues injectableValues = new InjectableValues()
    {
        @Override
        public Object findInjectableValue(Object valueId,DeserializationContext ctxt,BeanProperty forProperty,Object beanInstance)
        {
            return null;
        }
    };
    final ConfigurationFactoryFactory<? extends Configuration> configurationFactoryFactory = bootstrap.getConfigurationFactoryFactory();
    ConfigurationFactoryFactory factoryFactory = new ConfigurationFactoryFactory()
    {
        @Override
        public ConfigurationFactory create(Class klass,Validator validator,ObjectMapper objectMapper,String propertyPrefix)
        {
            objectMapper.setInjectableValues(injectableValues);
            //noinspection unchecked
            return configurationFactoryFactory.create(klass,propertyPrefix);
        }
    };
    //noinspection unchecked
    bootstrap.setConfigurationFactoryFactory(factoryFactory);
}

@Test
public void testParsing() throws Exception {
    ConfigurationFactory<MacroBaseConf> cfFactory = new ConfigurationFactory<>(MacroBaseConf.class,null,Jackson.newObjectMapper(),"");
    MacroBaseConf conf = cfFactory.build(new File("src/test/resources/conf/simple.yaml"));


    assertEquals((Double) 0.1,conf.getDouble("this.is.a.double"));
    assertEquals((Integer) 100,conf.getInt("this.is.an.integer"));
    assertEquals((Long) 10000000000000L,conf.getLong("this.is.a.long"));
    assertEquals("Test",conf.getString("this.is.a.string"));

    List<String> stringList = Lists.newArrayList("T1","T2","T3","T4");
    assertArrayEquals(stringList.toArray(),conf.getStringList("this.is.a.stringList").toArray());
    assertArrayEquals(stringList.toArray(),conf.getStringList("this.is.a.stringList.without.spaces").toArray());
    assertArrayEquals(stringList.toArray(),conf.getStringList("this.is.a.stringList.with.mixed.spaces").toArray());
}

/**
 * Builds a Configuration object from the file path given. It uses the {@link io.dropwizard.configuration.ConfigurationFactory}
 * to build the configuration.
 *
 * @param possibleFilename The path to the configuration.
 * @return A configuration object loaded form the filename given.
 */
private Config buildFromFile(String possibleFilename) {
    File configFile = new File(possibleFilename);
    Preconditions.checkArgument(configFile.exists(),"File must exist at: " + configFile.getAbsolutePath());

    try {
        return new ConfigurationFactory<>(
                configClass,Validation.buildDefaultValidatorFactory().getValidator(),"graceland")
                .build(configFile);

    } catch (Exception e) {
        String msg = "UnkNown exception triggered when attempting to build config from file:" + "\n" +
                "\t* Configuration Class: " + configClass.getCanonicalName() + "\n" +
                "\t* File: " + configFile.getAbsolutePath();

        throw new RuntimeException(msg,e);
    }
}

public static OregamiConfiguration createConfiguration(String configFilename) {
    ConfigurationFactory<OregamiConfiguration> factory =
            new ConfigurationFactory<>(
                    OregamiConfiguration.class,""
            );
    OregamiConfiguration configuration;
    try {
        configuration = factory.build(new File(configFilename));
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    System.out.println(ToStringBuilder.reflectionToString(configuration,ToStringStyle.MULTI_LINE_STYLE));
    System.out.println(ToStringBuilder.reflectionToString(configuration.getDatabaseConfiguration(),ToStringStyle.MULTI_LINE_STYLE));
    return configuration;
}

public SupportedMsaVersions loadSupportedMsaVersions(final URL url) {
    final ConfigurationFactory<SupportedMsaVersions> factory = supportedMsaVersionsFactoryFactory.create(
            SupportedMsaVersions.class,buildDefaultValidatorFactory().getValidator(),"");
    try {
        SupportedMsaVersions supportedMsaVersions = factory.build(
                configurationSourceProvider,url.toString());
        return supportedMsaVersions;
    } catch (IOException | ConfigurationException e) {
        throw propagate(e);
    }
}

public FileBackedConfigDataSource(
        ConfigConfiguration configuration,ConfigurationFactory<T> configurationFactory,String dataDirectory) {

    this.configuration = configuration;
    this.configurationFactory = configurationFactory;
    this.dataDirectory = dataDirectory;
}

@Before
public void beforeEach() throws Exception {
  factory = new ConfigurationFactory<>(
    MutableSchedulerConfiguration.class,Jackson.newObjectMapper().registerModule(new GuavaModule())
      .registerModule(new Jdk8Module()),"dw");

  configuration = factory.build(
    new SubstitutingSourceProvider(
      new FileConfigurationSourceProvider(),new EnvironmentvariableSubstitutor(false,true)),Resources.getResource("scheduler.yml").getFile());
}

private EmoConfiguration loadConfigFile(File configFile)
        throws IOException,ConfigurationException {
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());
    ConfigurationFactory<EmoConfiguration> configurationFactory = new ConfigurationFactory(EmoConfiguration.class,mapper,"dw");
    return configurationFactory.build(configFile);
}

protected EmoConfiguration getConfigurationFromresource()  throws Exception {
    URL url = BaseRoleConnectHelper.class.getResource(_configFileResource);
    Preconditions.checkNotNull(url,_configFileResource);
    File file = new File (url.toURI());
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());

    ConfigurationFactory<EmoConfiguration> configFactory = new ConfigurationFactory<>(EmoConfiguration.class,"dw");
    return configFactory.build(file);
}

@Test
public void ensureSdkDefaultConfigDeserialization()
        throws IOException,URISyntaxException,ConfigurationException {
    // This test makes sure that we haven't forgotten to update the emodb sdk default config file when we add/remove properties
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper mapper = CustomJsonObjectMapperFactory.build(new YAMLFactory());
    ConfigurationFactory configurationFactory = new ConfigurationFactory(EmoConfiguration.class,"dw");
    // Make sure that our config files are up to date
    configurationFactory.build(
    new File(EmoStartMojo.class.getResource("/emodb-default-config.yaml").toURI()));
}

@Override
public ConfigurationFactory<T> create(
    final Class<T> klass,final Validator validator,final ObjectMapper objectMapper,final String propertyPrefix
) {
    return new ImportableConfigurationFactory<T>(
        klass,configureObjectMapper(objectMapper.copy()),propertyPrefix
    );
}

/**
 * Parses the given configuration file and returns a configuration object.
 *
 * @param configurationFileName The name of the configuration file.
 * @return A configuration object.
 * @throws IOException The IO error that contains detail information.
 * @throws ConfigurationException The configuration error that contains detail information.
 */
public static ApiConfiguration parse(String configurationFileName)
  throws IOException,ConfigurationException {
  if (StringUtils.isBlank(configurationFileName)) {
    throw new IllegalArgumentException("Configuration file cannot be blank");
  }

  ObjectMapper objectMapper = null;
  if (configurationFileName.endsWith("yml") || configurationFileName.endsWith("yaml")) {
    objectMapper = Jackson.newObjectMapper(new YAMLFactory());
  } else if (configurationFileName.endsWith("json")) {
    objectMapper = Jackson.newObjectMapper(new JsonFactory());
  } else {
    throw new IllegalArgumentException("Unrecognized configuration file type");
  }

  ValidatorFactory validatorFactory = Validation
      .byProvider(HibernateValidator.class)
      .configure()
      .addValidatedValueHandler(new OptionalValidatedValueUnwrapper())
      .buildValidatorFactory();

  final ConfigurationFactory<ApiConfiguration> configurationFactory =
      new DefaultConfigurationFactoryFactory<ApiConfiguration>()
        .create(ApiConfiguration.class,"dw");

  final File file = new File(configurationFileName);
  if (!file.exists()) {
    throw new FileNotFoundException("Configuration file " + configurationFileName + " not found");
  }

  return configurationFactory.build(file);
}

@Test public void parsesLDAPLookupCorrectly() throws Exception {
  File yamlFile = new File(Resources.getResource("fixtures/keywhiz-ldap-lookup-test.yaml").getFile());
  Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
  ObjectMapper objectMapper = bootstrap.getobjectMapper().copy();

  LdapLookupConfig lookupConfig =
      new ConfigurationFactory<>(LdapLookupConfig.class,"dw")
          .build(yamlFile);

  assertthat(lookupConfig.getrequiredRoles()).containsOnly("keywhizAdmins");
  assertthat(lookupConfig.getRoleBasedn()).isEqualTo("ou=ApplicationAccess,dc=test,dc=com");
  assertthat(lookupConfig.getUserBasedn()).isEqualTo("ou=people,dc=com");
  assertthat(lookupConfig.getUserAttribute()).isEqualTo("uid");
}

static Injector createInjector() {
  KeywhizService service = new KeywhizService();
  Bootstrap<KeywhizConfig> bootstrap = new Bootstrap<>(service);
  service.initialize(bootstrap);

  File yamlFile = new File(Resources.getResource("keywhiz-test.yaml").getFile());
  Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
  ObjectMapper objectMapper = bootstrap.getobjectMapper().copy();
  KeywhizConfig config;
  try {
    config = new ConfigurationFactory<>(KeywhizConfig.class,"dw")
        .build(yamlFile);
  } catch (IOException | ConfigurationException e) {
    throw Throwables.propagate(e);
  }

  Environment environment = new Environment(service.getName(),bootstrap.getMetricRegistry(),bootstrap.getClassLoader());

  Injector injector = Guice.createInjector(new ServiceModule(config,environment));

  service.setInjector(injector);
  return injector;
}

@Test
public void testViaDW() throws Exception
{
    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    ObjectMapper objectMapper = Jackson.newObjectMapper();
    ConfigurationFactory<MyConfiguration> configurationFactory = new DefaultConfigurationFactoryFactory<MyConfiguration>().create(MyConfiguration.class,"dw");
    MyConfiguration configuration = configurationFactory.build(new FlexibleConfigurationSourceProvider(),"%{\"testValue\": \"override\"}");
    Assert.assertEquals("override",configuration.testValue);
}

public AdminConsoleAppBuilder<T> withConfigurationClass(final Class<T> configurationClass)
{
    factoryFactory = new ConfigurationFactoryFactory<T>()
    {
        @Override
        public ConfigurationFactory<T> create(Class<T> klass,String propertyPrefix)
        {
            return new YamlConfigurationFactory<>(configurationClass,propertyPrefix);
        }
    };
    return this;
}

@Before
public void setuptest() throws Exception {
    azureTableConfiguration = new ConfigurationFactory<>(
            AzureTableConfiguration.class,"dw.").build(new File(Resources.getResource("azure-test.yml").toURI()));
}

@Override
public ConfigurationFactory<T> create(
        Class<T>     klass,Validator    validator,String       propertyPrefix) {
    return new HoconConfigurationFactory<>(klass,propertyPrefix);
}

@Test
public void verifyConfigurable() throws Exception {
    ObjectMapper mapper = Jackson.newObjectMapper();

    // dropwizard 0.9.1 changed the validation wiring a bit..
    Class<ValidatedValueUnwrapper> optValidatorClazz = (Class<ValidatedValueUnwrapper>) Class
            .forName("io.dropwizard.validation.valuehandling.OptionalValidatedValueUnwrapper");

    Validator validator = Validation.buildDefaultValidatorFactory().getValidator();
    if (optValidatorClazz != null) {
        validator = Validation.byProvider(HibernateValidator.class).configure()
                .addValidatedValueHandler(optValidatorClazz.newInstance())
                .buildValidatorFactory().getValidator();
    }

    ConfigurationFactory<CloudWatchReporterFactory> configFactory =
            new ConfigurationFactory<>(CloudWatchReporterFactory.class,"dw");
    CloudWatchReporterFactory f = configFactory.build(new File(Resources.getResource("cw.yml").getFile()));

    assertEquals("[env=default]",f.getGlobalDimensions().toString());
    assertEquals("us-east-1",f.getAwsRegion());
    assertEquals("a.b",f.getNamespace());
    assertEquals("XXXXX",f.getAwsSecretKey());
    assertEquals("11111",f.getAwsAccessKeyId());
    assertEquals("p.neustar.biz",f.getAwsClientConfiguration().getProxyHost());
    assertNull(f.getAwsClientConfiguration().getProxyUsername());
}

public static PersistenceConfig createConfiguration(File configFile) {
  ConfigurationFactory<PersistenceConfig> factory =
      new ConfigurationFactory<>(PersistenceConfig.class,"");
  PersistenceConfig configuration;
  try {
    configuration = factory.build(configFile);
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
  return configuration;
}

/**
 * Returns a configuration object read in from the {@code fileName}.
 */
protected <T extends Configuration> T getConfiguration(String filename,Class<T> configurationClass) throws Exception {
  final ConfigurationFactory<T> configurationFactory = new ConfigurationFactory<>(
      configurationClass,"dw");
  if (filename != null) {
    final File file = new File(Resources.getResource(filename).getFile());
    if (!file.exists())
      throw new FileNotFoundException("File " + file + " not found");
    return configurationFactory.build(file);
  }

  return configurationFactory.build();
}

@Before
public void beforeEach() throws Exception {
    server = new TestingServer();

    server.start();

    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,Jackson.newObjectMapper().registerModule(
                            new GuavaModule())
                            .registerModule(new Jdk8Module()),"dw");

    config = factory.build(
            new SubstitutingSourceProvider(
                    new FileConfigurationSourceProvider(),Resources.getResource("scheduler.yml").getFile());

    ServiceConfig initial = config.createConfig().getServiceConfig();

    final CassandraSchedulerConfiguration targetConfig = config.createConfig();
    clusterTaskConfig = targetConfig.getClusterTaskConfig();

    final CuratorFrameworkConfig curatorConfig = config.getCuratorConfig();
    RetryPolicy retryPolicy =
            (curatorConfig.getoperationTimeout().isPresent()) ?
                    new RetryUntilElapsed(
                            curatorConfig.getoperationTimeoutMs()
                                    .get()
                                    .intValue(),(int) curatorConfig.getBackoffMs()) :
                    new RetryForever((int) curatorConfig.getBackoffMs());

    stateStore = new CuratorStateStore(
            targetConfig.getServiceConfig().getName(),server.getConnectString(),retryPolicy);
    stateStore.storeFrameworkId(Protos.FrameworkID.newBuilder().setValue("1234").build());
    identity = new IdentityManager(
            initial,stateStore);

    identity.register("test_id");

    DefaultConfigurationManager configurationManager =
            new DefaultConfigurationManager(CassandraSchedulerConfiguration.class,config.createConfig().getServiceConfig().getName(),config.createConfig(),new ConfigValidator(),stateStore);

    Capabilities mockCapabilities = Mockito.mock(Capabilities.class);
    when(mockCapabilities.supportsNamedVips()).thenReturn(true);
    configuration = new ConfigurationManager(
            new CassandraDaemonTask.Factory(mockCapabilities),configurationManager);

    cassandraState = new CassandraState(
            configuration,clusterTaskConfig,stateStore);
}

@BeforeClass
public static void beforeAll() throws Exception {

    server = new TestingServer();

    server.start();

    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,Resources.getResource("scheduler.yml").getFile());

    final CuratorFrameworkConfig curatorConfig = config.getCuratorConfig();
    RetryPolicy retryPolicy =
            (curatorConfig.getoperationTimeout().isPresent()) ?
                    new RetryUntilElapsed(
                            curatorConfig.getoperationTimeoutMs()
                                    .get()
                                    .intValue(),(int) curatorConfig.getBackoffMs()) :
                    new RetryForever((int) curatorConfig.getBackoffMs());

    stateStore = new CuratorStateStore(
            config.createConfig().getServiceConfig().getName(),retryPolicy);

    final CassandraSchedulerConfiguration configuration = config.createConfig();
    try {
        final ConfigValidator configValidator = new ConfigValidator();
        final DefaultConfigurationManager defaultConfigurationManager =
                new DefaultConfigurationManager(CassandraSchedulerConfiguration.class,configuration.getServiceConfig().getName(),configuration,configValidator,stateStore);
        Capabilities mockCapabilities = Mockito.mock(Capabilities.class);
        when(mockCapabilities.supportsNamedVips()).thenReturn(true);
        configurationManager = new ConfigurationManager(
                new CassandraDaemonTask.Factory(mockCapabilities),defaultConfigurationManager);
    } catch (ConfigStoreException e) {
        throw new RuntimeException(e);
    }
}

@BeforeClass
public static void beforeAll() throws Exception {
    server = new TestingServer();
    server.start();
    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,"dw");
    MutableSchedulerConfiguration mutable = factory.build(
            new SubstitutingSourceProvider(
                    new FileConfigurationSourceProvider(),Resources.getResource("scheduler.yml").getFile());
    config = mutable.createConfig();

    final CuratorFrameworkConfig curatorConfig = mutable.getCuratorConfig();
    RetryPolicy retryPolicy =
            (curatorConfig.getoperationTimeout().isPresent()) ?
                    new RetryUntilElapsed(
                            curatorConfig.getoperationTimeoutMs()
                                    .get()
                                    .intValue(),(int) curatorConfig.getBackoffMs()) :
                    new RetryForever((int) curatorConfig.getBackoffMs());

    StateStore stateStore = new CuratorStateStore(
            config.getServiceConfig().getName(),retryPolicy);
    configurationManager =
            new DefaultConfigurationManager(CassandraSchedulerConfiguration.class,config.getServiceConfig().getName(),config,stateStore);
    config = (CassandraSchedulerConfiguration) configurationManager.getTargetConfig();
}

private void beforeHelper(String configName) throws Exception {
    MockitoAnnotations.initMocks(this);
    mesosConfig = Mockito.mock(MesosConfig.class);

    client = Mockito.mock(SchedulerClient.class);
    Mockito.when(mockFuture.get()).thenReturn(true);
    Mockito.when(mockStage.toCompletableFuture()).thenReturn(mockFuture);
    backup = Mockito.mock(backupmanager.class);
    restore = Mockito.mock(RestoreManager.class);
    cleanup = Mockito.mock(CleanupManager.class);
    repair = Mockito.mock(RepairManager.class);
    upgrade = Mockito.mock(UpgradesstableManager.class);
    seeds = Mockito.mock(SeedsManager.class);
    capabilities = Mockito.mock(Capabilities.class);

    executorService = Executors.newSingleThreadScheduledExecutor();
    frameworkId = TestUtils.generateFrameworkId();

    factory = new ConfigurationFactory<>(
            MutableSchedulerConfiguration.class,Jackson.newObjectMapper().registerModule(
                    new GuavaModule())
                    .registerModule(new Jdk8Module()),Resources.getResource(configName).getFile());

    stateStore = new CuratorStateStore(
            "/" + config.getServiceConfig().getName(),server.getConnectString());

    DefaultConfigurationManager defaultConfigurationManager
            = new DefaultConfigurationManager(CassandraSchedulerConfiguration.class,"/" + config.createConfig().getServiceConfig().getName(),stateStore);


    Capabilities mockCapabilities = Mockito.mock(Capabilities.class);
    when(mockCapabilities.supportsNamedVips()).thenReturn(true);
    configurationManager = new ConfigurationManager(
            new CassandraDaemonTask.Factory(mockCapabilities),defaultConfigurationManager);
    cassandraState = new CassandraState(
            configurationManager,configurationManager.getTargetConfig().getClusterTaskConfig(),stateStore);

    offerRequirementProvider = new PersistentOfferRequirementProvider(defaultConfigurationManager);
    scheduler = new CassandraScheduler(
            configurationManager,mesosConfig,offerRequirementProvider,cassandraState,client,backup,restore,cleanup,repair,upgrade,true,seeds,executorService,stateStore,defaultConfigurationManager,capabilities);

    masterInfo = TestUtils.generateMasterInfo();

    driver = new QueuedSchedulerDriver();
    scheduler.setSchedulerDriver(driver);
    scheduler.registered(driver,frameworkId,masterInfo);
}

@Before
public void beforeEach() throws Exception {
    MockitoAnnotations.initMocks(this);
    server = new TestingServer();
    server.start();

    Capabilities mockCapabilities = mock(Capabilities.class);
    when(mockCapabilities.supportsNamedVips()).thenReturn(true);
    taskFactory = new CassandraDaemonTask.Factory(mockCapabilities);

    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,Resources.getResource("scheduler.yml").getFile());

    final CassandraSchedulerConfiguration targetConfig = config.createConfig();
    clusterTaskConfig = targetConfig.getClusterTaskConfig();

    final CuratorFrameworkConfig curatorConfig = config.getCuratorConfig();
    RetryPolicy retryPolicy =
            (curatorConfig.getoperationTimeout().isPresent()) ?
                    new RetryUntilElapsed(
                            curatorConfig.getoperationTimeoutMs()
                                    .get()
                                    .intValue(),retryPolicy);
    stateStore.storeFrameworkId(Protos.FrameworkID.newBuilder().setValue("1234").build());

    configurationManager = new DefaultConfigurationManager(CassandraSchedulerConfiguration.class,stateStore);

    cassandraState = new CassandraState(
            new ConfigurationManager(taskFactory,configurationManager),stateStore);
}

@BeforeClass
public static void beforeAll() throws Exception {

    server = new TestingServer();

    server.start();

    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,"dw");

    MutableSchedulerConfiguration mutable = factory.build(
            new SubstitutingSourceProvider(
                    new FileConfigurationSourceProvider(),Resources.getResource("scheduler.yml").getFile());

    config = mutable.createConfig();
    ServiceConfig initial = config.getServiceConfig();

    clusterTaskConfig = config.getClusterTaskConfig();

    final CuratorFrameworkConfig curatorConfig = mutable.getCuratorConfig();
    RetryPolicy retryPolicy =
            (curatorConfig.getoperationTimeout().isPresent()) ?
                    new RetryUntilElapsed(
                            curatorConfig.getoperationTimeoutMs()
                                    .get()
                                    .intValue(),(int) curatorConfig.getBackoffMs()) :
                    new RetryForever((int) curatorConfig.getBackoffMs());

    stateStore = new CuratorStateStore(
            config.getServiceConfig().getName(),retryPolicy);
    stateStore.storeFrameworkId(Protos.FrameworkID.newBuilder().setValue("1234").build());

    identity = new IdentityManager(
            initial,stateStore);
    Capabilities mockCapabilities = Mockito.mock(Capabilities.class);
    when(mockCapabilities.supportsNamedVips()).thenReturn(true);
    configuration = new ConfigurationManager(
            new CassandraDaemonTask.Factory(mockCapabilities),configurationManager);

    provider = new ClusterTaskOfferRequirementProvider();
}

@Before
public void beforeEach() throws Exception {
    MockitoAnnotations.initMocks(this);
    server = new TestingServer();
    server.start();

    final ConfigurationFactory<MutableSchedulerConfiguration> factory =
            new ConfigurationFactory<>(
                    MutableSchedulerConfiguration.class,retryPolicy);
    stateStore.storeFrameworkId(Protos.FrameworkID.newBuilder().setValue("1234").build());
    identity = new IdentityManager(initial,stateStore);

    taskFactory = new CassandrataskFactory(executorDriver);
}

public static void main(String... args) throws Exception {
    final String DROPWIZARD_PROPERTY_PREFIX = "dw";

    // Load the config.yaml file specified as the first argument.
    ConfigurationFactory<EmoConfiguration> configFactory = new ConfigurationFactory(
            EmoConfiguration.class,DROPWIZARD_PROPERTY_PREFIX);
    EmoConfiguration configuration = configFactory.build(new File(args[0]));
    int numWriterThreads = Integer.parseInt(args[1]);
    int numReaderThreads = Integer.parseInt(args[2]);
    String adminApiKey = configuration.getAuthorizationConfiguration().getAdminApiKey();

    MetricRegistry metricRegistry = new MetricRegistry();
    new LoggingFactory().configure(metricRegistry,"stress");

    CuratorFramework curator = configuration.getZooKeeperConfiguration().newCurator();
    curator.start();

    QueueClientFactory queueFactory = QueueClientFactory.forClusterandHttpConfiguration(
            configuration.getCluster(),configuration.getHttpClientConfiguration(),metricRegistry);
    AuthQueueService authQueueService = ServicePoolBuilder.create(AuthQueueService.class)
            .withServiceFactory(queueFactory)
            .withHostdiscovery(new ZooKeeperHostdiscovery(curator,queueFactory.getServiceName(),metricRegistry))
            .withMetricRegistry(metricRegistry)
            .withCachingPolicy(ServiceCachingPolicyBuilder.getMultiThreadedClientPolicy())
            .buildProxy(new ExponentialBackoffRetry(5,50,1000,TimeUnit.MILLISECONDS));
    QueueService queueService = QueueServiceAuthenticator.proxied(authQueueService)
            .usingCredentials(adminApiKey);

    final QueueStresstest stresstest = new QueueStresstest(queueService);

    ThreadFactory writerFactory = new ThreadFactoryBuilder().setNameFormat("Writer-%d").build();
    for (int i = 0; i < numWriterThreads; i++) {
        writerFactory.newThread(new Runnable() {
            @Override
            public void run() {
                stresstest.write();
            }
        }).start();
    }

    ThreadFactory readerFactory = new ThreadFactoryBuilder().setNameFormat("Reader-%d").build();
    for (int i = 0; i < numReaderThreads; i++) {
        readerFactory.newThread(new Runnable() {
            @Override
            public void run() {
                stresstest.read();
            }
        }).start();
    }

    ThreadFactory reportFactory = new ThreadFactoryBuilder().setNameFormat("Report-%d").build();
    Executors.newScheduledThreadPool(1,reportFactory).scheduleAtFixedrate(new Runnable() {
        @Override
        public void run() {
            stresstest.report();
        }
    },1,TimeUnit.SECONDS);

    // Run forever
}

public static void main(String args[]) throws Exception {
    ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
    // Load the config.yaml file specified as the first argument.
    ConfigurationFactory<EmoConfiguration> configFactory = new ConfigurationFactory<>(
            EmoConfiguration.class,"dw");
    File configFile = new File(args[0]);
    EmoConfiguration configuration = configFactory.build(configFile);
    String ddlFilePath = args[1];
    File scanUploadDir = new File(args[2]);

    checkArgument(scanUploadDir.isDirectory(),"Not a valid directory: %s",scanUploadDir);
    checkArgument(configuration.getServiceMode() == EmoServiceMode.SCANNER,"Not configured for scanner: %s",configuration.getServiceMode());

    // To prevent conflicting with EmoDB running on this same server adjust the host and admin ports.
    updatePortsToAvoidCollision(configuration.getServerFactory());

    HostAndPort hostAndPort = new SelfHostAndPortModule().provideSelfHostAndPort(configuration.getServerFactory());
    MetricRegistry metricRegistry = new MetricRegistry();
    configuration.getLoggingFactory().configure(metricRegistry,"scan");

    DataStore dataStore = null;

    try (CuratorFramework curator = configuration.getZooKeeperConfiguration().newCurator()) {
        curator.start();

        DataStoreClientFactory dataStoreFactory = DataStoreClientFactory.forClusterandHttpConfiguration(
                configuration.getCluster(),metricRegistry);
        dataStore = ServicePoolBuilder.create(DataStore.class)
                .withServiceFactory(dataStoreFactory.usingCredentials(configuration.getScanner().get().getScannerApiKey().get()))
                .withHostdiscovery(new ZooKeeperHostdiscovery(curator,dataStoreFactory.getServiceName(),metricRegistry))
                .withCachingPolicy(ServiceCachingPolicyBuilder.getMultiThreadedClientPolicy())
                .withMetricRegistry(metricRegistry)
                .buildProxy(new ExponentialBackoffRetry(5,TimeUnit.MILLISECONDS));

        ScanUploadTest test = new ScanUploadTest(dataStore);
        test.createAndPopulateTestTables();
        test.scanToDirectory(configuration,ddlFilePath,configFile,scanUploadDir,hostAndPort,validatorFactory,metricRegistry);
        test.validateScanResults(scanUploadDir);
    } finally {
        if (dataStore != null) {
            ServicePoolProxies.close(dataStore);
        }
    }
}

public static void main(String... args) throws Exception {
    final String DROPWIZARD_PROPERTY_PREFIX = "dw";

    // Load the config.yaml file specified as the first argument.
    ConfigurationFactory<EmoConfiguration> configFactory = new ConfigurationFactory(
            EmoConfiguration.class,DROPWIZARD_PROPERTY_PREFIX);
    EmoConfiguration configuration = configFactory.build(new File(args[0]));
    int numWriterThreads = Integer.parseInt(args[1]);
    int numReaderThreads = Integer.parseInt(args[2]);
    String apiKey = configuration.getAuthorizationConfiguration().getAdminApiKey();

    MetricRegistry metricRegistry = new MetricRegistry();
    new LoggingFactory().configure(metricRegistry,"stress");

    CuratorFramework curator = configuration.getZooKeeperConfiguration().newCurator();
    curator.start();

    DataStoreClientFactory dataStoreFactory = DataStoreClientFactory.forClusterandHttpConfiguration(
            configuration.getCluster(),metricRegistry);
    AuthDataStore authDataStore = ServicePoolBuilder.create(AuthDataStore.class)
            .withServiceFactory(dataStoreFactory)
            .withHostdiscovery(new ZooKeeperHostdiscovery(curator,TimeUnit.MILLISECONDS));
    DataStore dataStore = DataStoreAuthenticator.proxied(authDataStore).usingCredentials(apiKey);

    DatabusClientFactory databusFactory = DatabusClientFactory.forClusterandHttpConfiguration(
            configuration.getCluster(),metricRegistry);
    AuthDatabus authDatabus = ServicePoolBuilder.create(AuthDatabus.class)
            .withServiceFactory(databusFactory)
            .withHostdiscovery(new ZooKeeperHostdiscovery(curator,databusFactory.getServiceName(),TimeUnit.MILLISECONDS));
    Databus databus = DatabusAuthenticator.proxied(authDatabus).usingCredentials(apiKey);

    final SorStresstest stresstest = new SorStresstest(dataStore,databus);

    if (!dataStore.getTableExists(TABLE)) {
        TableOptions options = new TableOptionsBuilder().setPlacement("ugc_global:ugc").build();
        dataStore.createTable(TABLE,options,ImmutableMap.of("table",TABLE),new AuditBuilder().setLocalHost().build());
    }

    databus.subscribe(SUBSCRIPTION,Conditions.alwaysTrue(),Duration.standardDays(7),Duration.standardDays(1));

    ThreadFactory writerFactory = new ThreadFactoryBuilder().setNameFormat("SoR Writer-%d").build();
    for (int i = 0; i < numWriterThreads; i++) {
        writerFactory.newThread(new Runnable() {
            @Override
            public void run() {
                stresstest.writeDeltas();
            }
        }).start();
    }

    ThreadFactory readerFactory = new ThreadFactoryBuilder().setNameFormat("Databus Reader-%d").build();
    for (int i = 0; i < numReaderThreads; i++) {
        readerFactory.newThread(new Runnable() {
            @Override
            public void run() {
                stresstest.readDatabus();
            }
        }).start();
    }

    ThreadFactory reportFactory = new ThreadFactoryBuilder().setNameFormat("Report-%d").build();
    Executors.newScheduledThreadPool(1,TimeUnit.SECONDS);

    ServicePoolProxies.close(dataStore);
    Closeables.close(curator,true);
}

/**
 * Creates a Jetty server for an application that can be started / stopped in-process
 *
 * @param config           An application configuration instance (with properties set)
 * @param applicationClass The {@link io.dropwizard.Application} implementation class
 * @param <T>              The configuration class
 * @return A Jetty server
 */
@SuppressWarnings("unchecked")
public static <T extends Configuration> DropWizardServer<T> createServer(
    T config,Class<? extends Application<T>> applicationClass) throws Exception {
  // Create application
  final Application<T> application = applicationClass.getConstructor().newInstance();

  // Create bootstrap
  final ServerCommand<T> serverCommand = new ServerCommand<T>(application);
  final Bootstrap<T> bootstrap = new Bootstrap<T>(application);
  bootstrap.addCommand(serverCommand);

  // Write a temporary config file
  File tmpConfigFile = new File(
      System.getProperty("java.io.tmpdir"),config.getClass().getCanonicalName() + "_" + System.currentTimeMillis());
  tmpConfigFile.deleteOnExit();
  bootstrap.getobjectMapper().writeValue(tmpConfigFile,config);

  // Parse configuration
  ConfigurationFactory<T> configurationFactory
      = bootstrap.getConfigurationFactoryFactory()
      .create((Class<T>) config.getClass(),bootstrap.getValidatorFactory().getValidator(),bootstrap.getobjectMapper(),"dw");
  final T builtConfig = configurationFactory.build(
      bootstrap.getConfigurationSourceProvider(),tmpConfigFile.getAbsolutePath());

  // Configure logging
  builtConfig.getLoggingFactory()
      .configure(bootstrap.getMetricRegistry(),bootstrap.getApplication().getName());

  // Environment
  final Environment environment = new Environment(bootstrap.getApplication().getName(),bootstrap.getClassLoader());

  // Initialize environment
  builtConfig.getMetricsFactory().configure(environment.lifecycle(),bootstrap.getMetricRegistry());

  // Server
  final Server server = builtConfig.getServerFactory().build(environment);
  server.addLifeCycleListener(new AbstractLifeCycle.AbstractLifeCycleListener() {
    @Override
    public void lifeCycleStopped(LifeCycle event) {
      builtConfig.getLoggingFactory().stop();
    }
  });

  return new DropWizardServer(builtConfig,bootstrap,application,environment,server,environment.metrics());
}

/**
 * Creates a Jetty server for an application that can be started / stopped in-process
 *
 * @param config
 *  An application configuration instance (with properties set)
 * @param applicationClass
 *  The {@link io.dropwizard.Application} implementation class
 * @param <T>
 *  The configuration class
 * @return
 *  A Jetty server
 */
@SuppressWarnings("unchecked")
public static <T extends Configuration>
DropWizardServer<T> createServer(T config,Class<? extends Application<T>> applicationClass) throws Exception
{
  // Create application
  final Application<T> application = applicationClass.getConstructor().newInstance();

  // Create bootstrap
  final ServerCommand<T> serverCommand = new ServerCommand<T>(application);
  final Bootstrap<T> bootstrap = new Bootstrap<T>(application);
  bootstrap.addCommand(serverCommand);

  // Write a temporary config file
  File tmpConfigFile = new File(
          System.getProperty("java.io.tmpdir"),config);

  // Parse configuration
  ConfigurationFactory<T> configurationFactory
          = bootstrap.getConfigurationFactoryFactory()
                     .create((Class<T>) config.getClass(),"dw");
  final T builtConfig = configurationFactory.build(
          bootstrap.getConfigurationSourceProvider(),tmpConfigFile.getAbsolutePath());

  // Configure logging
  builtConfig.getLoggingFactory()
             .configure(bootstrap.getMetricRegistry(),bootstrap.getMetricRegistry());

  // Server
  final Server server = builtConfig.getServerFactory().build(environment);
  server.addLifeCycleListener(new AbstractLifeCycle.AbstractLifeCycleListener()
  {
    @Override
    public void lifeCycleStopped(LifeCycle event)
    {
      builtConfig.getLoggingFactory().stop();
    }
  });

  return new DropWizardServer(builtConfig,environment.metrics());
}

@Override
public ConfigurationFactory<T> get() {
  return factory;
}

@Override
public ConfigurationFactory create(final Class klass,final String propertyPrefix) {
    return new EnvironmentConfigurationFactory<>(klass,propertyPrefix,new DefaultEnvironmentProvider());
}