public static AbstractConfiguration createDynamicConfig() {
  LOGGER.info("create dynamic config:");
  ConcurrentCompositeConfiguration config = ConfigUtil.createLocalConfig();
  DynamicWatchedConfiguration configFromConfigCenter = createConfigFromConfigCenter(config);
  if (configFromConfigCenter != null) {
    ConcurrentMapConfiguration injectConfig = new ConcurrentMapConfiguration();
    config.addConfigurationAtFront(injectConfig,"extraInjectConfig");

    duplicateServiceCombConfigToCse(configFromConfigCenter);
    config.addConfigurationAtFront(configFromConfigCenter,"configCenterConfig");

    configFromConfigCenter.getSource().addUpdateListener(new ServiceCombPropertyUpdateListener(injectConfig));
  }

  return config;
}

@Test
public void testCreateDynamicConfigNoConfigCenterSPI() {
  new Expectations(SPIServiceUtils.class) {
    {
      SPIServiceUtils.getTargetService(ConfigCenterConfigurationSource.class);
      result = null;
    }
  };

  AbstractConfiguration dynamicConfig = ConfigUtil.createDynamicConfig();
  MicroserviceConfigLoader loader = ConfigUtil.getMicroserviceConfigLoader(dynamicConfig);
  List<ConfigModel> list = loader.getConfigModels();
  Assert.assertEquals(loader,ConfigUtil.getMicroserviceConfigLoader(dynamicConfig));
  Assert.assertEquals(1,list.size());
  Assert.assertNotEquals(DynamicWatchedConfiguration.class,((ConcurrentCompositeConfiguration) dynamicConfig).getConfiguration(0).getClass());
}

@BeforeClass
public static void beforeCls() {
  ConfigUtil.installDynamicConfig();
  AbstractConfiguration configuration =
      (AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource();
  configuration.addProperty("cse.loadbalance.test.transactionControl.policy","org.apache.servicecomb.loadbalance.filter.SimpleTransactionControlFilter");
  configuration.addProperty("cse.loadbalance.test.transactionControl.options.tag0","value0");
  configuration.addProperty("cse.loadbalance.test.isolation.enabled","true");
  configuration.addProperty("cse.loadbalance.serverListFilters","a");
  configuration.addProperty("cse.loadbalance.serverListFilter.a.className","org.apache.servicecomb.loadbalance.MyServerListFilterExt");
}

@Before
public void setUp() throws Exception {
  IsolationServerListFilter = new IsolationServerListFilter();
  loadBalancerStats = new LoadBalancerStats("loadBalancer");

  AbstractConfiguration configuration =
      (AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource();
  configuration.clearProperty("cse.loadbalance.isolation.enabled");
  configuration.addProperty("cse.loadbalance.isolation.enabled","true");
  configuration.clearProperty("cse.loadbalance.isolation.enableRequestThreshold");
  configuration.addProperty("cse.loadbalance.isolation.enableRequestThreshold","3");
}

@After
public void tearDown() throws Exception {
  IsolationServerListFilter = null;
  loadBalancerStats = null;

  AbstractConfiguration configuration =
      (AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource();
  configuration.clearProperty("cse.loadbalance.isolation.continuousFailureThreshold");
}

@Test
public void testGetFilteredlistofServersOnContinuousFailureReachesThreshold() {
  ((AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource())
      .addProperty("cse.loadbalance.isolation.continuousFailureThreshold","3");
  Invocation invocation = Mockito.mock(Invocation.class);
  CseServer testServer = Mockito.mock(CseServer.class);
  Mockito.when(invocation.getMicroserviceName()).thenReturn("microserviceName");
  Mockito.when(testServer.getCountinuousFailureCount()).thenReturn(3);
  Mockito.when(testServer.getLastVisitTime()).thenReturn(System.currentTimeMillis());

  for (int i = 0; i < 3; ++i) {
    loadBalancerStats.incrementNumRequests(testServer);
  }

  List<Server> serverList = new ArrayList<>();
  serverList.add(testServer);
  IsolationServerListFilter.setLoadBalancerStats(loadBalancerStats);
  IsolationServerListFilter.setInvocation(invocation);
  List<Server> returnedServerList = IsolationServerListFilter.getFilteredlistofServers(serverList);
  Assert.assertEquals(0,returnedServerList.size());
}

@Test
public void testGetFilteredlistofServersOnContinuousFailureIsBelowThreshold() {
  ((AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource())
      .addProperty("cse.loadbalance.isolation.continuousFailureThreshold","3");
  Invocation invocation = Mockito.mock(Invocation.class);
  CseServer testServer = Mockito.mock(CseServer.class);
  Mockito.when(invocation.getMicroserviceName()).thenReturn("microserviceName");
  Mockito.when(testServer.getCountinuousFailureCount()).thenReturn(2);
  Mockito.when(testServer.getLastVisitTime()).thenReturn(System.currentTimeMillis());

  for (int i = 0; i < 3; ++i) {
    loadBalancerStats.incrementNumRequests(testServer);
  }

  List<Server> serverList = new ArrayList<>();
  serverList.add(testServer);
  IsolationServerListFilter.setLoadBalancerStats(loadBalancerStats);
  IsolationServerListFilter.setInvocation(invocation);
  List<Server> returnedServerList = IsolationServerListFilter.getFilteredlistofServers(serverList);
  Assert.assertEquals(1,returnedServerList.size());
}

@BeforeClass
public static void beforeCls() {
  ConfigUtil.installDynamicConfig();
  AbstractConfiguration configuration =
      (AbstractConfiguration) DynamicPropertyFactory.getbackingConfigurationSource();
  configuration.addProperty(REQUEST_TIMEOUT_KEY,2000);
}

@Override
public void init(AbstractConfiguration config,ApplicationListenerFactory factory) {
    init(config);

    logger.trace("Initializing Kafka consumer ...");

    // consumer config
    Properties props = new Properties();
    props.put("bootstrap.servers",config.getString("bootstrap.servers"));
    props.put("group.id",config.getString("group.id"));
    props.put("enable.auto.commit","true");
    props.put("key.serializer",StringSerializer.class.getName());
    props.put("value.serializer",InternalMessageSerializer.class.getName());

    // consumer
    this.consumer = new KafkaConsumer<>(props);

    // consumer worker
    this.worker = new KafkaApplicationWorker(this.consumer,APPLICATION_TOPIC,factory.newListener());
    this.executor.submit(this.worker);
}

@Override
public void init(AbstractConfiguration config,String brokerId,brokerListenerFactory factory) {
    init(config);

    broKER_TOPIC = broKER_TOPIC_PREFIX + "." + brokerId;

    logger.trace("Initializing Kafka consumer ...");

    // consumer config
    Properties props = new Properties();
    props.put("bootstrap.servers",UUIDs.shortUuid());
    props.put("enable.auto.commit",InternalMessageSerializer.class.getName());

    // consumer
    this.consumer = new KafkaConsumer<>(props);

    // consumer worker
    this.worker = new KafkabrokerWorker(this.consumer,broKER_TOPIC,factory.newListener());
    this.executor.submit(this.worker);
}

protected void init(AbstractConfiguration config) {
    broKER_TOPIC_PREFIX = config.getString("communicator.broker.topic");
    APPLICATION_TOPIC = config.getString("communicator.application.topic");

    logger.trace("Initializing Kafka producer ...");

    // producer config
    Properties props = new Properties();
    props.put("bootstrap.servers",config.getString("bootstrap.servers"));
    props.put("acks",config.getString("acks"));
    props.put("key.serializer",InternalMessageSerializer.class.getName());

    // producer
    this.producer = new KafkaProducer<>(props);

    // consumer executor
    this.executor = Executors.newSingleThreadExecutor();
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("single")) {
        throw new IllegalStateException("RedisSyncSingleStorage class can only be used with single redis setup,but redis.type value is " + config.getString("redis.type"));
    }

    List<String> address = parseRedisAddress(config.getString("redis.address"),6379);
    int databaseNumber = config.getInt("redis.database",0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuce = RedisClient.create(lettuceURI);
    this.lettuceConn = this.lettuce.connect();

    // redisson
    Config redissonConfig = new Config();
    redissonConfig.useSingleServer()
            .setAddress(address.get(0))
            .setDatabase(databaseNumber)
            .setPassword(StringUtils.isNotEmpty(password) ? password : null);
    this.redisson = Redisson.create(redissonConfig);

    // params
    initParams(config);
}

@Override
public void init(AbstractConfiguration config,String serverId) {
    try {
        ConnectionFactory cf = new ConnectionFactory();
        cf.setUsername(config.getString("rabbitmq.userName",ConnectionFactory.DEFAULT_USER));
        cf.setPassword(config.getString("rabbitmq.password",ConnectionFactory.DEFAULT_PASS));
        cf.setVirtualHost(config.getString("rabbitmq.virtualHost",ConnectionFactory.DEFAULT_VHOST));
        cf.setAutomaticRecoveryEnabled(true);
        cf.setExceptionHandler(new RabbitMQExceptionHandler());
        this.conn = cf.newConnection(Address.parseAddresses(config.getString("rabbitmq.addresses")));
        this.channel = conn.createChannel();

        logger.trace("Initializing RabbitMQ broker resources ...");

        broKER_TOPIC_PREFIX = config.getString("communicator.broker.topic");

        logger.trace("Initializing RabbitMQ application resources ...");

        APPLICATION_TOPIC = config.getString("communicator.application.topic");
        this.channel.exchangeDeclare(APPLICATION_TOPIC,"topic",true);

    } catch (IOException | TimeoutException e) {
        logger.error("Failed to connect to RabbitMQ servers",e);
        throw new IllegalStateException("Init RabbitMQ communicator Failed");
    }
}

/**
 * Attempts to acquire the Vault URL from Archaius.
 *
 * @return Vault URL
 */
@Nullable
@Override
public  String resolve() {
    final AbstractConfiguration configuration = ConfigurationManager.getConfigInstance();
    final String envUrl = configuration.getString(CERBERUS_ADDR_ENV_PROPERTY);
    final String sysUrl = configuration.getString(CERBERUS_ADDR_SYS_PROPERTY);

    if (StringUtils.isNotBlank(envUrl) && HttpUrl.parse(envUrl) != null) {
        return envUrl;
    } else if (StringUtils.isNotBlank(sysUrl) && HttpUrl.parse(sysUrl) != null) {
        return sysUrl;
    }

    logger.warn("Unable to resolve the Cerberus URL.");

    return null;
}

@Override
public Config createConfig(String name) {
  if (CONfig != null) {
    return CONfig;
  }
  synchronized (ArchaiusBaseFactory.class) {
    if (CONfig == null) {
      AbstractConfiguration configuration = getConfiguration();
      ConfigurationManager.install(configuration);
      CONfig = new ArchaiusWrapper(configuration);
      ConfigFactory.setContext(CONfig);
      configuration.addConfigurationListener(event -> {
        if (!event.isBeforeUpdate()) {
          CONfig.fire(event.getPropertyName());
        }
      });
    }
  }
  return CONfig;
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("single")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with single redis setup,0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuce = RedisClient.create(lettuceURI);
    this.lettuceConn = this.lettuce.connect();

    // params
    initParams(config);
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("sentinel")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with sentinel redis setup,26379);
    int databaseNumber = config.getInt("redis.database",0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";
    String masterId = config.getString("redis.master");

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis-sentinel://" + password + String.join(",",address) + "/" + databaseNumber + "#" + masterId);
    this.lettuceSentinel = RedisClient.create(lettuceURI);
    this.lettuceSentinelConn = MasterSlave.connect(this.lettuceSentinel,new Utf8StringCodec(),lettuceURI);
    this.lettuceSentinelConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // params
    initParams(config);
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("master_slave")) {
        throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with master slave redis setup,0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuceMasterSlave = RedisClient.create(lettuceURI);
    this.lettuceMasterSlaveConn = MasterSlave.connect(this.lettuceMasterSlave,lettuceURI);
    this.lettuceMasterSlaveConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // params
    initParams(config);
}

/** Create a new instance. */
@Inject
Plugin(Registry registry) throws IOException {

  AbstractConfiguration config = ConfigurationManager.getConfigInstance();
  final boolean enabled = config.getBoolean(ENABLED_PROP,true);
  if (enabled) {
    ConfigurationManager.loadPropertiesFromresources(CONfig_FILE);
    if (config.getBoolean("spectator.gc.loggingEnabled")) {
      GC_LOGGER.start(null);
      LOGGER.info("gc logging started");
    } else {
      LOGGER.info("gc logging is not enabled");
    }

    Jmx.registerStandardMXBeans(registry);
  } else {
    LOGGER.debug("plugin not enabled,set " + ENABLED_PROP + "=true to enable");
  }
}

@Override
protected void doGet(HttpServletRequest req,HttpServletResponse resp) throws servletexception,IOException {
    // get list of properties
    TreeSet<String> properties = new TreeSet<String>();
    AbstractConfiguration config = ConfigurationManager.getConfigInstance();
    Iterator<String> keys = config.getKeys();
    while (keys.hasNext()) {
        String key = keys.next();
        Object value = config.getProperty(key);
        if ("aws.accessId".equals(key)
                || "aws.secretKey".equals(key)
                || "experiments-service.secret".equals(key)
                || "java.class.path".equals(key)
                || key.contains("framework.securityDeFinition")
                || key.contains("password")
                || key.contains("secret")) {
            value = "*****";
        }
        properties.add(key + "=" + value.toString());
    }

    // write them out in sorted order
    for (String line : properties) {
        resp.getWriter().append(line).println();
    }
}

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
    logger.debug("Received ContextRefreshedEvent {}",event);

    if (event.getSource().equals(getBootstrapApplicationContext())) {
        //the root context is fully started
        appMetadata = bootstrapApplicationContext.getBean(AppMetadata.class);
        configuration = bootstrapApplicationContext.getBean(AbstractConfiguration.class);
        configurationProvider = bootstrapApplicationContext.getBean(ConfigurationProvider.class);

        logger.debug("Root context started");

        initClientApplication();

        return;
    }

    if (event.getSource() instanceof ApplicationContext && ((ApplicationContext) event.getSource()).getId().equals(appMetadata.getName())) {
        //the child context is fully started
        this.applicationContext = (AbstractApplicationContext) event.getSource();

        logger.debug("Child context started");
    }

    state.compareAndSet(State.STARTING,State.RUNNING);
}

@Bean
@SuppressWarnings("resource")
public AbstractConfiguration applicationConfiguration() throws ClassNotFoundException {
    AppMetadata appMetadata = appMetadata();
    ServerInstanceContext serverInstanceContext = serverInstanceContext();
    if(appEnvironment == null && serverInstanceContext != null){
        appEnvironment = serverInstanceContext.getEnvironment();
    }
    ConfigurationBuilder configurationBuilder = new ConfigurationBuilder(appMetadata.getName(),appEnvironment,addSystemConfigs,reflections());
    configurationBuilder.withConfigurationProvider(configurationProvider());
    configurationBuilder.withServerInstanceContext(serverInstanceContext());
    configurationBuilder.withApplicationProperties(appMetadata.getPropertiesResourceLocation());
    configurationBuilder.withScanModuleConfigurations(scanModuleConfigurations);
    configurationBuilder.withAppVersion(appMetadata.getDeclaringClass().getPackage().getImplementationVersion());
    AbstractConfiguration configuration = configurationBuilder.build();
    if(serverInstanceContext != null){
        serverInstanceContext.setAppName(appMetadata.getName());
        serverInstanceContext.setVersion(configuration.getString(BootstrapConfigKeys.APP_VERSION_KEY.getPropertyName()));
    }
    return configuration;
}

@Override
public void onApplicationEvent(EnvironmentChangeEvent event) {
    AbstractConfiguration manager = ConfigurationManager.getConfigInstance();
    for (String key : event.getKeys()) {
        for (ConfigurationListener listener : manager
                .getConfigurationListeners()) {
            Object source = event.getSource();
            // Todo: Handle add vs set vs delete?
            int type = AbstractConfiguration.EVENT_SET_PROPERTY;
            String value = this.env.getProperty(key);
            boolean beforeUpdate = false;
            listener.configurationChanged(new ConfigurationEvent(source,type,key,value,beforeUpdate));
        }
    }
}

private static void addArchaiusConfiguration(ConcurrentCompositeConfiguration config) {
    if (ConfigurationManager.isConfigurationInstalled()) {
        AbstractConfiguration installedConfiguration = ConfigurationManager
                .getConfigInstance();
        if (installedConfiguration instanceof ConcurrentCompositeConfiguration) {
            ConcurrentCompositeConfiguration configInstance = (ConcurrentCompositeConfiguration) installedConfiguration;
            configInstance.addConfiguration(config);
        }
        else {
            installedConfiguration.append(config);
            if (!(installedConfiguration instanceof AggregatedConfiguration)) {
                log.warn(
                        "Appending a configuration to an existing non-aggregated installed configuration will have no effect");
            }
        }
    }
    else {
        ConfigurationManager.install(config);
    }
}

@Test
public void testSunnyDayNoClientAuth() throws Exception{

    AbstractConfiguration cm = ConfigurationManager.getConfigInstance();

    String name = "GetPostSecureTest" + ".testSunnyDayNoClientAuth";

    String configPrefix = name + "." + "ribbon";

    cm.setProperty(configPrefix + "." + CommonClientConfigKey.IsSecure,"true");
    cm.setProperty(configPrefix + "." + CommonClientConfigKey.SecurePort,Integer.toString(PORT2));
    cm.setProperty(configPrefix + "." + CommonClientConfigKey.IsHostnameValidationrequired,"false");
    cm.setProperty(configPrefix + "." + CommonClientConfigKey.TrustStore,FILE_TS2.getAbsolutePath());
    cm.setProperty(configPrefix + "." + CommonClientConfigKey.TrustStorePassword,PASSWORD);

    RestClient rc = (RestClient) ClientFactory.getNamedClient(name);

    testServer2.accept();

    URI getUri = new URI(SERVICE_URI2 + "test/");
       HttpRequest request = HttpRequest.newBuilder().uri(getUri).queryParams("name","test").build();
    HttpResponse response = rc.execute(request);
    assertEquals(200,response.getStatus());
}

@Before
public void setUp() {
  AbstractConfiguration.setDefaultListDelimiter(',');
  clearTestSystemProperties();
  this.configurationHelper = new ConfigurationHelper();
  this.test1Properties = new HashMap<String,String>() {{
      this.put("a.b.c","efgh");
      this.put("a.b.d","1234");
    }};

  this.test3Properties = new HashMap<String,"jklm");
      this.put("e.f.h","90123");
      // The value in the file is "foo,bar" but AbstractConfiguration.getString(key) only returns
      // the first item in a collection.
      this.put("i.j.k","foo");
    }};
}

private void invokeListeners()
{
    if (m_configurationListeners != null)
    {
        try
        {
            ConfigurationEvent event = new ConfigurationEvent(this,AbstractConfiguration.EVENT_SET_PROPERTY,null,this,false);
            for (ConfigurationListener listener:m_configurationListeners)
            {
                listener.configurationChanged(event);
            }
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
    }
}

public static AbstractConfiguration convertEnvVariable(AbstractConfiguration source) {
  Iterator<String> keys = source.getKeys();
  while (keys.hasNext()) {
    String key = keys.next();
    String[] separatedKey = key.split(CONfig_KEY_SPLITER);
    if (separatedKey.length == 1) {
      continue;
    }
    String newKey = String.join(".",separatedKey);
    source.addProperty(newKey,source.getProperty(key));
  }
  return source;
}

private static void duplicateServiceCombConfigToCse(AbstractConfiguration source) {
  Iterator<String> keys = source.getKeys();
  while (keys.hasNext()) {
    String key = keys.next();
    if (!key.startsWith(CONfig_SERVICECOMB_PREFIX)) {
      continue;
    }

    String cseKey = CONfig_CSE_PREFIX + key.substring(key.indexOf(".") + 1);
    source.addProperty(cseKey,source.getProperty(key));
  }
}

private static void duplicateServiceCombConfigToCse(ConcurrentCompositeConfiguration compositeConfiguration,AbstractConfiguration source,String sourceName) {
  duplicateServiceCombConfigToCse(source);

  compositeConfiguration.addConfiguration(source,sourceName);
}

public static void installDynamicConfig() {
  if (ConfigurationManager.isConfigurationInstalled()) {
    LOGGER.warn("Configuration installed by others,will ignore this configuration.");
    return;
  }

  AbstractConfiguration dynamicConfig = ConfigUtil.createDynamicConfig();
  ConfigurationManager.install(dynamicConfig);
}

@Override
protected Properties mergeProperties() throws IOException {
  Properties properties = super.mergeProperties();

  AbstractConfiguration config = ConfigurationManager.getConfigInstance();
  Iterator<String> iter = config.getKeys();
  while (iter.hasNext()) {
    String key = iter.next();
    Object value = config.getProperty(key);
    properties.put(key,value);
  }
  return properties;
}

@Test
public void testCreateDynamicConfigHasConfigCenter(
    @Mocked ConfigCenterConfigurationSource configCenterConfigurationSource) {
  AbstractConfiguration dynamicConfig = ConfigUtil.createDynamicConfig();
  Assert.assertEquals(DynamicWatchedConfiguration.class,((ConcurrentCompositeConfiguration) dynamicConfig).getConfiguration(0).getClass());
}

@Test
public void duplicateServiceCombConfigToCseListValue() throws Exception {
  List<String> list = Arrays.asList("a","b");

  AbstractConfiguration config = new DynamicConfiguration();
  config.addProperty("servicecomb.list",list);
  Deencapsulation.invoke(ConfigUtil.class,"duplicateServiceCombConfigToCse",config);

  Object result = config.getProperty("cse.list");
  assertthat(result,instanceOf(List.class));
  assertthat(result,equalTo(list));
}

@Test
public void testConvertEnvVariable() {
  String someProperty = "cse_service_registry_address";
  AbstractConfiguration config = new DynamicConfiguration();
  config.addProperty(someProperty,"testing");
  AbstractConfiguration result = ConfigUtil.convertEnvVariable(config);
  assertthat(result.getString("cse.service.registry.address"),equalTo("testing"));
  assertthat(result.getString("cse_service_registry_address"),equalTo("testing"));
}

@Test
public void testCreateMicroserviceInstanceFromFile() {
  AbstractConfiguration config = ConfigUtil.createDynamicConfig();
  ConcurrentCompositeConfiguration configuration = new ConcurrentCompositeConfiguration();
  configuration.addConfiguration(config);
  ConfigurationManager.install(configuration);
  MicroserviceInstance instance = MicroserviceInstance.createFromDeFinition(config);
  Assert.assertEquals(instance.getDataCenterInfo().getName(),"myDC");
  Assert.assertEquals(instance.getDataCenterInfo().getRegion(),"my-Region");
  Assert.assertEquals(instance.getDataCenterInfo().getAvailableZone(),"my-Zone");
}

@BeforeClass
public static void initSetup() throws Exception {
  AbstractConfiguration dynamicConfig = ConfigUtil.createDynamicConfig();
  ConcurrentCompositeConfiguration configuration = new ConcurrentCompositeConfiguration();
  configuration.addConfiguration(dynamicConfig);
  configuration.addConfiguration(inMemoryConfig);

  ConfigurationManager.install(configuration);
}

@BeforeClass
public static void beforeCls() {
  AbstractConfiguration configuration = new BaseConfiguration();
  configuration.addProperty("cse.loadbalance.test.flowsplitFilter.policy","org.apache.servicecomb.loadbalance.filter.SimpleFlowsplitFilter");
  configuration.addProperty("cse.loadbalance.test.flowsplitFilter.options.tag0","value0");
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("master_slave")) {
        throw new IllegalStateException("RedisSyncSingleStorage class can only be used with master slave redis setup,lettuceURI);
    this.lettuceMasterSlaveConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // redisson
    String masterNode = address.get(0);
    String[] slaveNodes = address.subList(1,address.size()).toArray(new String[address.size() - 1]);
    Config redissonConfig = new Config();
    redissonConfig.useMasterSlaveServers()
            .setMasteraddress(masterNode)
            .setLoadBalancer(new RoundRobinLoadBalancer())
            .addSlaveAddress(slaveNodes)
            .setReadMode(ReadMode.MASTER)
            .setDatabase(databaseNumber)
            .setPassword(StringUtils.isNotEmpty(password) ? password : null);
    this.redisson = Redisson.create(redissonConfig);

    // params
    initParams(config);
}

@Override
public void init(AbstractConfiguration config) {
    if (!config.getString("redis.type").equals("cluster")) {
        throw new IllegalStateException("RedisSyncSingleStorage class can only be used with cluster redis setup,0);
    String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

    // lettuce
    RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
    this.lettuceCluster = RedisClusterClient.create(lettuceURI);
    this.lettuceCluster.setoptions(new ClusterClientOptions.Builder()
            .refreshClusterView(true)
            .refreshPeriod(1,TimeUnit.MINUTES)
            .build());
    this.lettuceClusterConn = this.lettuceCluster.connect();
    this.lettuceClusterConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

    // redisson
    Config redissonConfig = new Config();
    redissonConfig.useClusterServers()
            .setScanInterval(60000)
            .addNodeAddress(address.toArray(new String[address.size()]))
            .setReadMode(ReadMode.MASTER)
            .setPassword(StringUtils.isNotEmpty(password) ? password : null);
    this.redisson = Redisson.create(redissonConfig);

    // params
    initParams(config);
}