RedisKeyValueStateProvider implements StateProvider { StateConfig getStateConfig(Map stormConf) throws Exception { StateConfig stateConfig = null; String providerConfig = null; ObjectMapper mapper = new ObjectMapper(); mapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); if (stormConf.containsKey(org.apache.storm.Config.TOPOLOGY_STATE_PROVIDER_CONFIG)) { providerConfig = (String) stormConf.get(org.apache.storm.Config.TOPOLOGY_STATE_PROVIDER_CONFIG); stateConfig = mapper.readValue(providerConfig, StateConfig.class); } else { stateConfig = new StateConfig(); } return stateConfig; }  @Override State newState(String namespace, Map stormConf, TopologyContext context);  }
@Test public void testgetDefaultConfig() throws Exception { RedisKeyValueStateProvider provider = new RedisKeyValueStateProvider(); Map<String, String> stormConf = new HashMap<>(); RedisKeyValueStateProvider.StateConfig config = provider.getStateConfig(stormConf); assertNotNull(config); }
@Test public void testgetConfigWithProviderConfig() throws Exception { RedisKeyValueStateProvider provider = new RedisKeyValueStateProvider(); Map<String, String> stormConf = new HashMap<>(); stormConf.put(Config.TOPOLOGY_STATE_PROVIDER_CONFIG, "{\"keyClass\":\"String\", \"valueClass\":\"String\"," + " \"jedisPoolConfig\":" + "{\"host\":\"localhost\", \"port\":1000}}"); RedisKeyValueStateProvider.StateConfig config = provider.getStateConfig(stormConf); assertEquals("String", config.keyClass); assertEquals("String", config.valueClass); assertEquals("localhost", config.jedisPoolConfig.getHost()); assertEquals(1000, config.jedisPoolConfig.getPort()); }