Config { private <T> T get(Map<String, String> map, Value<T> value) { if (!CONFIG_VALUES.containsKey(value.key)) { throw new IllegalArgumentException("Unknown config value: " + value.key + " probably needs to be added to Config.CONFIG_VALUES"); } final String s = map.getOrDefault(value.key, value.defaultValue); if (s != null) { return (T) value.type.parse(s); } else { if (value.required) { throw new IllegalArgumentException("Config value: " + value.key + " is mandatory"); } return null; } } Config(Map<String, String> map); static Collection<Value<?>> keys(); static Set<String> keyNames(); @SuppressWarnings("unchecked") T get(Value<T> value, T defaultValue); @SuppressWarnings("unchecked") T get(Value<T> value); @Override boolean equals(Object o); @Override int hashCode(); static final String TC_RESOURCE_LABELS; static final String TC_KAFKA_BOOTSTRAP_SERVERS; static final String TC_NAMESPACE; static final String TC_ZK_CONNECT; static final String TC_ZK_SESSION_TIMEOUT_MS; static final String TC_ZK_CONNECTION_TIMEOUT_MS; static final String TC_PERIODIC_INTERVAL_MS; static final String TC_REASSIGN_THROTTLE; static final String TC_REASSIGN_VERIFY_INTERVAL_MS; static final String TC_TOPIC_METADATA_MAX_ATTEMPTS; static final String TC_TOPICS_PATH; static final String TC_TLS_ENABLED; static final String TC_TLS_TRUSTSTORE_LOCATION; static final String TC_TLS_TRUSTSTORE_PASSWORD; static final String TC_TLS_KEYSTORE_LOCATION; static final String TC_TLS_KEYSTORE_PASSWORD; static final String TC_TLS_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM; static final Value<Labels> LABELS; static final Value<String> KAFKA_BOOTSTRAP_SERVERS; static final Value<String> NAMESPACE; static final Value<String> ZOOKEEPER_CONNECT; static final Value<Long> ZOOKEEPER_SESSION_TIMEOUT_MS; static final Value<Long> ZOOKEEPER_CONNECTION_TIMEOUT_MS; static final Value<Long> FULL_RECONCILIATION_INTERVAL_MS; static final Value<Long> REASSIGN_THROTTLE; static final Value<Long> REASSIGN_VERIFY_INTERVAL_MS; static final Value<Integer> TOPIC_METADATA_MAX_ATTEMPTS; static final Value<String> TOPICS_PATH; static final Value<String> TLS_ENABLED; static final Value<String> TLS_TRUSTSTORE_LOCATION; static final Value<String> TLS_TRUSTSTORE_PASSWORD; static final Value<String> TLS_KEYSTORE_LOCATION; static final Value<String> TLS_KEYSTORE_PASSWORD; static final Value<String> TLS_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM; }
@Test public void testDefaultInput() { Map<String, String> map = new HashMap<>(MANDATORY); Config c = new Config(map); assertThat(c.get(Config.ZOOKEEPER_SESSION_TIMEOUT_MS).intValue(), is(20_000)); }
@Test public void testOverrideZookeeperSessionTimeout() { Map<String, String> map = new HashMap<>(MANDATORY); map.put(Config.ZOOKEEPER_SESSION_TIMEOUT_MS.key, "13000"); Config c = new Config(map); assertThat(c.get(Config.ZOOKEEPER_SESSION_TIMEOUT_MS).intValue(), is(13_000)); }
@Test public void testTopicMetadataMaxAttemptsIsSetCorrectly() { Map<String, String> map = new HashMap<>(MANDATORY); map.put(Config.TC_TOPIC_METADATA_MAX_ATTEMPTS, "3"); Config c = new Config(map); assertThat(c.get(Config.TOPIC_METADATA_MAX_ATTEMPTS).intValue(), is(3)); }