[GitHub] [cassandra] dcapwell commented on a diff in pull request #2133: CASSANDRA-15254 Support update statements over the SettingsTable virtual table

[via GitHub]

dcapwell commented on code in PR #2133:
URL: https://github.com/apache/cassandra/pull/2133#discussion_r1095171823


##########
src/java/org/apache/cassandra/config/DatabaseDescriptor.java:
##########
@@ -3711,7 +3718,13 @@ public static boolean getCDCBlockWrites()
         return conf.cdc_block_writes;
     }
 
+    @Deprecated

Review Comment:
   why?  This isn't a public class, we can make breaking changes here



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();

Review Comment:
   why do we need locks?



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));

Review Comment:
   you shouldn't hold the ref, this can change on you causing this class to 
drift.  `DD` is what owns `Config`, we should reference it and not cache



##########
src/java/org/apache/cassandra/service/StorageService.java:
##########
@@ -384,6 +386,8 @@ public StorageService()
         jmxObjectName = "org.apache.cassandra.db:type=StorageService";
 
         sstablesTracker = new 
SSTablesGlobalTracker(SSTableFormat.Type.current());
+
+        
ConfigurationRegistry.instance.addPropertyChangeListener(REPAIR_REQUEST_TIMEOUT,
 this::repairRequestTimeoutListener, DurationSpec.LongMillisecondsBound.class);

Review Comment:
   this doesn't need to exist... All "mutators" via JMX log the mutation, this 
may be in DD or in the Mean itself (such as this case), there is no reason to 
trigger this as all this does is update `Config`.
   
   the vtable can produce a similar log, so don't need to duplicate



##########
src/java/org/apache/cassandra/db/virtual/SettingsTable.java:
##########
@@ -59,102 +56,83 @@ final class SettingsTable extends AbstractVirtualTable
                            .addPartitionKeyColumn(NAME, UTF8Type.instance)
                            .addRegularColumn(VALUE, UTF8Type.instance)
                            .build());
-        this.config = config;
+        registerConverters();
+        this.configurationRegistry = configurationRegistry;
     }
 
-    @Override
-    public DataSet data(DecoratedKey partitionKey)
+    private void registerConverters()
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        String name = UTF8Type.instance.compose(partitionKey.getKey());
-        if (BACKWARDS_COMPATABLE_NAMES.containsKey(name))
-            ClientWarn.instance.warn("key '" + name + "' is deprecated; should 
switch to '" + BACKWARDS_COMPATABLE_NAMES.get(name) + "'");
-        if (PROPERTIES.containsKey(name))
-            result.row(name).column(VALUE, getValue(PROPERTIES.get(name)));
-        return result;
+        converterRegistry.put(Boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(Integer.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(int.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(DurationSpec.LongMillisecondsBound.class,
+                              
(PropertyConverter<DurationSpec.LongMillisecondsBound>) value -> 
DurationSpec.from(value, DurationSpec.LongMillisecondsBound.class));
     }
 
-    @Override
-    public DataSet data()
+    /**
+     * Setter for the property.
+     * @param name the name of the property.
+     * @param value the string representation of the value of the property to 
set.
+     */
+    private void setProperty(String name, String value)
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        for (Map.Entry<String, Property> e : PROPERTIES.entrySet())
-            result.row(e.getKey()).column(VALUE, getValue(e.getValue()));
-        return result;
+        Class<?> setterType = configurationRegistry.getPropertyType(name);
+        PropertyConverter<?> converter = converterRegistry.get(setterType);
+        if (converter == null)
+            throw invalidRequest("Unknown converter for property with name 
'%s' and type '%s'", name, setterType);
+
+        configurationRegistry.update(name, value == null ? null : 
converter.convert(value));
     }
 
-    private String getValue(Property prop)
+    private static @Nullable String getProperty(ConfigurationRegistry 
registry, String name)
     {
-        Object value = prop.get(config);
+        Object value = registry.get(name);
         return value == null ? null : value.toString();
     }
 
-    private static Map<String, Property> getProperties()
+    @Override
+    protected void applyColumnDeletion(ColumnValues partitionKey, ColumnValues 
clusteringColumns, String columnName)
     {
-        Loader loader = Properties.defaultLoader();
-        Map<String, Property> properties = loader.flatten(Config.class);
-        // only handling top-level replacements for now, previous logic was 
only top level so not a regression
-        Map<String, Replacement> replacements = 
Replacements.getNameReplacements(Config.class).get(Config.class);
-        if (replacements != null)
-        {
-            for (Replacement r : replacements.values())
-            {
-                Property latest = properties.get(r.newName);
-                assert latest != null : "Unable to find replacement new name: 
" + r.newName;
-                Property conflict = properties.put(r.oldName, 
r.toProperty(latest));
-                // some configs kept the same name, but changed the type, if 
this is detected then rely on the replaced property
-                assert conflict == null || r.oldName.equals(r.newName) : 
String.format("New property %s attempted to replace %s, but this property 
already exists", latest.getName(), conflict.getName());
-            }
-        }
-        for (Map.Entry<String, String> e : 
BACKWARDS_COMPATABLE_NAMES.entrySet())
-        {
-            String oldName = e.getKey();
-            if (properties.containsKey(oldName))
-                throw new AssertionError("Name " + oldName + " is present in 
Config, this adds a conflict as this name had a different meaning in " + 
SettingsTable.class.getSimpleName());
-            String newName = e.getValue();
-            Property prop = Objects.requireNonNull(properties.get(newName), 
newName + " cant be found for " + oldName);
-            properties.put(oldName, Properties.rename(oldName, prop));
-        }
-        return properties;
+        String key = partitionKey.value(0);
+        setProperty(key, null);
     }
 
-    /**
-     * settings table was released in 4.0 and attempted to support nested 
properties for a few hand selected properties.
-     * The issue is that 4.0 used '_' to seperate the names, which makes it 
hard to map back to the yaml names; to solve
-     * this 4.1+ uses '.' to avoid possible conflicts, this class provides 
mappings from old names to the '.' names.
-     *
-     * There were a handle full of properties which had custom names, names 
not present in the yaml, this map also
-     * fixes this and returns the proper (what is accessable via yaml) names.
-     */
-    private static Map<String, String> getBackwardsCompatableNames()
+    @Override
+    protected void applyColumnUpdate(ColumnValues partitionKey,
+                                     ColumnValues clusteringColumns,
+                                     Optional<ColumnValue> columnValue)
     {
-        Map<String, String> names = new HashMap<>();
-        // Names that dont match yaml
-        names.put("audit_logging_options_logger", 
"audit_logging_options.logger.class_name");
-        names.put("server_encryption_options_client_auth", 
"server_encryption_options.require_client_auth");
-        names.put("server_encryption_options_endpoint_verification", 
"server_encryption_options.require_endpoint_verification");
-        names.put("server_encryption_options_legacy_ssl_storage_port", 
"server_encryption_options.legacy_ssl_storage_port_enabled");
-        names.put("server_encryption_options_protocol", 
"server_encryption_options.accepted_protocols");
-
-        // matching names
-        names.put("audit_logging_options_audit_logs_dir", 
"audit_logging_options.audit_logs_dir");
-        names.put("audit_logging_options_enabled", 
"audit_logging_options.enabled");
-        names.put("audit_logging_options_excluded_categories", 
"audit_logging_options.excluded_categories");
-        names.put("audit_logging_options_excluded_keyspaces", 
"audit_logging_options.excluded_keyspaces");
-        names.put("audit_logging_options_excluded_users", 
"audit_logging_options.excluded_users");
-        names.put("audit_logging_options_included_categories", 
"audit_logging_options.included_categories");
-        names.put("audit_logging_options_included_keyspaces", 
"audit_logging_options.included_keyspaces");
-        names.put("audit_logging_options_included_users", 
"audit_logging_options.included_users");
-        names.put("server_encryption_options_algorithm", 
"server_encryption_options.algorithm");
-        names.put("server_encryption_options_cipher_suites", 
"server_encryption_options.cipher_suites");
-        names.put("server_encryption_options_enabled", 
"server_encryption_options.enabled");
-        names.put("server_encryption_options_internode_encryption", 
"server_encryption_options.internode_encryption");
-        names.put("server_encryption_options_optional", 
"server_encryption_options.optional");
-        names.put("transparent_data_encryption_options_chunk_length_kb", 
"transparent_data_encryption_options.chunk_length_kb");
-        names.put("transparent_data_encryption_options_cipher", 
"transparent_data_encryption_options.cipher");
-        names.put("transparent_data_encryption_options_enabled", 
"transparent_data_encryption_options.enabled");
-        names.put("transparent_data_encryption_options_iv_length", 
"transparent_data_encryption_options.iv_length");
-
-        return names;

Review Comment:
   why are you removing?  These strings only make sense to this class, so 
pulling out to a config registry is not safe, only this class should know about 
config names it created in the past



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>

Review Comment:
   should *not* extend `AbstractMap<String, Object>`, this can be error prone 
as there are 2 other maps and from an API everything takes a `String`, so what 
does `ConfigurationRegistry.get` actually mean?  What about 
`ConfigurationRegistry.put` or `ConfigurationRegistry.removeKey`?



##########
src/java/org/apache/cassandra/config/DurationSpec.java:
##########
@@ -135,6 +135,25 @@ public TimeUnit unit()
         return unit;
     }
 
+    public static <T extends DurationSpec> T from(String value, Class<T> clazz)

Review Comment:
   can we remove?  The only usage can be replaced with 
`DurationSpec.IntMillisecondsBound::new`



##########
src/java/org/apache/cassandra/config/registry/PropertyChangeListener.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+/**
+ * Interface for listening to configuration property changes.
+ */
+public interface PropertyChangeListener<T>
+{
+    /**
+     * Called before a configuration property has been changed.
+     *
+     * @param name the name of the property.
+     * @param oldValue the old value of the property.
+     * @param newValue the new value of the property.
+     */
+    void onBeforeChange(String name, T oldValue, T newValue);

Review Comment:
   based off name, should there also be `onAfterChange`?



##########
src/java/org/apache/cassandra/db/commitlog/CommitLog.java:
##########
@@ -122,6 +135,9 @@ private static CommitLog construct()
 
         // register metrics
         metrics.attach(executor, segmentManager);
+
+        
ConfigurationRegistry.instance.addPropertyChangeListener(CDC_BLOCK_WRITES, 
this::cdcBlockWritesChangeListener, Boolean.class);

Review Comment:
   Looks like the majority of complexity of this patch is to add listeners for 
this single listener?  There maybe more cases as you flesh this out, but at 
least in the current PR this is the only real listener (the others just log you 
mutated, so don't need to listen).
   
   I feel we could simplify to
   
   ```ConfigurationRegistry.instance.register("cdc_block_writes", this:: 
setCDCBlockWrites0)```
   
   Where `setCDCBlockWrites0` is just the logic in 
`cdcBlockWritesChangeListener`
   
   I don't like the `Boolean.class` as it's brittle and can drift; in fact this 
patch provides the wrong type!  You want `Boolean.TYPE`.
   



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();
+        try
+        {
+            Object oldValue = unit.getValue();
+            if (listeners.get(name) != null)
+                listeners.get(name).onBeforeChange(name, oldValue, value);
+
+            unit.setValue(value);
+            // This potentially may expose the values that are not safe to see 
in logs on production.
+            logger.info("Updated property {} from {} to {}", name, oldValue, 
value);
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Error updating 
property with name '%s', cause: %s", name, e.getMessage()), e);
+        }
+        finally
+        {
+            unit.lock.unlock();
+        }
+    }
+
+    public Class<?> getPropertyType(String key)
+    {
+        if (!registry.containsKey(key))
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", key));
+
+        return registry.get(key).getter.getType();
+    }
+
+    public String getBackwardCompatibleKey(String key)
+    {
+        return BACKWARDS_COMPATABLE_NAMES.get(key);
+    }
+
+    /**
+     * Adds a listener for the property with the given name.
+     * @param name property name to listen to.
+     * @param listener listener to add.
+     * @param <T> type of the property.
+     */
+    public <T> void addPropertyChangeListener(String name, 
PropertyChangeListener<T> listener, Class<T> listenerType)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe for resitering handler.", name));
+
+        if (unit.setterType == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+//        Class<T> type = (Class<T>) ((ParameterizedType) 
listener.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
+        if (!unit.setterType.equals(listenerType))
+            throw new ConfigurationException(String.format("Property with name 
'%s' expects type '%s', but got '%s'.", name, unit.setterType, listenerType));
+
+        listeners.put(name, new PropertyChangeListenerWrapper<>(listener, 
listenerType));
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public Set<Entry<String, Object>> entrySet()
+    {
+        return transformSet;
+    }
+
+    private static Map<String, Property> getProperties()
+    {
+        Loader loader = Properties.defaultLoader();
+
+        Map<String, Property> properties = loader.flatten(Config.class);
+        // only handling top-level replacements for now, previous logic was 
only top level so not a regression
+        Map<String, Replacement> replacements = 
Replacements.getNameReplacements(Config.class).get(Config.class);
+        if (replacements != null)
+        {
+            for (Replacement r : replacements.values())
+            {
+                Property latest = properties.get(r.newName);
+                assert latest != null : "Unable to find replacement new name: 
" + r.newName;
+                Property conflict = properties.put(r.oldName, 
r.toProperty(latest));
+                // some configs kept the same name, but changed the type, if 
this is detected then rely on the replaced property
+                assert conflict == null || r.oldName.equals(r.newName) : 
String.format("New property %s attempted to replace %s, but this property 
already exists", latest.getName(), conflict.getName());
+            }
+        }
+
+        for (Map.Entry<String, String> e : 
BACKWARDS_COMPATABLE_NAMES.entrySet())
+        {
+            String oldName = e.getKey();
+            if (properties.containsKey(oldName))
+                throw new AssertionError("Name " + oldName + " is present in 
Config, this adds a conflict as this name had a different meaning.");
+            String newName = e.getValue();
+            Property prop = Objects.requireNonNull(properties.get(newName), 
newName + " cant be found for " + oldName);
+            properties.put(oldName, Properties.rename(oldName, prop));
+        }
+        return properties;
+    }
+
+    /**
+     * settings table was released in 4.0 and attempted to support nested 
properties for a few hand selected properties.
+     * The issue is that 4.0 used '_' to seperate the names, which makes it 
hard to map back to the yaml names; to solve
+     * this 4.1+ uses '.' to avoid possible conflicts, this class provides 
mappings from old names to the '.' names.
+     *
+     * There were a handle full of properties which had custom names, names 
not present in the yaml, this map also
+     * fixes this and returns the proper (what is accessable via yaml) names.
+     */
+    private static Map<String, String> getBackwardsCompatableNames()
+    {
+        Map<String, String> names = new HashMap<>();
+        // Names that don't match yaml
+        names.put("audit_logging_options_logger", 
"audit_logging_options.logger.class_name");
+        names.put("server_encryption_options_client_auth", 
"server_encryption_options.require_client_auth");
+        names.put("server_encryption_options_endpoint_verification", 
"server_encryption_options.require_endpoint_verification");
+        names.put("server_encryption_options_legacy_ssl_storage_port", 
"server_encryption_options.legacy_ssl_storage_port_enabled");
+        names.put("server_encryption_options_protocol", 
"server_encryption_options.accepted_protocols");
+
+        // matching names
+        names.put("audit_logging_options_audit_logs_dir", 
"audit_logging_options.audit_logs_dir");
+        names.put("audit_logging_options_enabled", 
"audit_logging_options.enabled");
+        names.put("audit_logging_options_excluded_categories", 
"audit_logging_options.excluded_categories");
+        names.put("audit_logging_options_excluded_keyspaces", 
"audit_logging_options.excluded_keyspaces");
+        names.put("audit_logging_options_excluded_users", 
"audit_logging_options.excluded_users");
+        names.put("audit_logging_options_included_categories", 
"audit_logging_options.included_categories");
+        names.put("audit_logging_options_included_keyspaces", 
"audit_logging_options.included_keyspaces");
+        names.put("audit_logging_options_included_users", 
"audit_logging_options.included_users");
+        names.put("server_encryption_options_algorithm", 
"server_encryption_options.algorithm");
+        names.put("server_encryption_options_cipher_suites", 
"server_encryption_options.cipher_suites");
+        names.put("server_encryption_options_enabled", 
"server_encryption_options.enabled");
+        names.put("server_encryption_options_internode_encryption", 
"server_encryption_options.internode_encryption");
+        names.put("server_encryption_options_optional", 
"server_encryption_options.optional");
+        names.put("transparent_data_encryption_options_chunk_length_kb", 
"transparent_data_encryption_options.chunk_length_kb");
+        names.put("transparent_data_encryption_options_cipher", 
"transparent_data_encryption_options.cipher");
+        names.put("transparent_data_encryption_options_enabled", 
"transparent_data_encryption_options.enabled");
+        names.put("transparent_data_encryption_options_iv_length", 
"transparent_data_encryption_options.iv_length");
+
+        return names;
+    }
+
+    private static class TransformSet<T, R> extends AbstractSet<R>
+    {
+        private final Set<T> set;
+        private final Function<T, R> transform;
+
+        public TransformSet(Set<T> set, Function<T, R> transform)
+        {
+            this.set = set;
+            this.transform = transform;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public Iterator<R> iterator()
+        {
+            Iterator<T> iter = set.iterator();
+            return new AbstractIterator<R>()
+            {
+                @Override
+                protected R computeNext()
+                {
+                    if (!iter.hasNext())
+                        return endOfData();
+
+                    T entry = iter.next();
+                    return transform.apply(entry);
+                }
+            };
+        }
+        /** {@inheritDoc} */
+        @Override
+        public int size()
+        {
+            return set.size();
+        }
+    }
+
+    private static class PropertyChangeListenerWrapper<T> implements 
PropertyChangeListener<T>
+    {
+        private final PropertyChangeListener<T> listener;
+        private final Class<T> type;
+
+        public PropertyChangeListenerWrapper(PropertyChangeListener<T> 
listener, Class<T> type)
+        {
+            this.listener = listener;
+            this.type = type;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public void onBeforeChange(String name, Object oldValue, Object 
newValue)
+        {
+            T castedOldValue = type.cast(oldValue);
+            T castedNewValue = type.cast(newValue);
+            listener.onBeforeChange(name, castedOldValue, castedNewValue);
+        }
+    }

Review Comment:
   is this needed?  If the user defines the call site as `boolean` the JVM will 
add the cost for you, so should be able to drop this whole class



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();
+        try
+        {
+            Object oldValue = unit.getValue();
+            if (listeners.get(name) != null)
+                listeners.get(name).onBeforeChange(name, oldValue, value);
+
+            unit.setValue(value);
+            // This potentially may expose the values that are not safe to see 
in logs on production.
+            logger.info("Updated property {} from {} to {}", name, oldValue, 
value);
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Error updating 
property with name '%s', cause: %s", name, e.getMessage()), e);
+        }
+        finally
+        {
+            unit.lock.unlock();
+        }
+    }
+
+    public Class<?> getPropertyType(String key)
+    {
+        if (!registry.containsKey(key))
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", key));
+
+        return registry.get(key).getter.getType();
+    }
+
+    public String getBackwardCompatibleKey(String key)
+    {
+        return BACKWARDS_COMPATABLE_NAMES.get(key);
+    }
+
+    /**
+     * Adds a listener for the property with the given name.
+     * @param name property name to listen to.
+     * @param listener listener to add.
+     * @param <T> type of the property.
+     */
+    public <T> void addPropertyChangeListener(String name, 
PropertyChangeListener<T> listener, Class<T> listenerType)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe for resitering handler.", name));
+
+        if (unit.setterType == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+//        Class<T> type = (Class<T>) ((ParameterizedType) 
listener.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
+        if (!unit.setterType.equals(listenerType))
+            throw new ConfigurationException(String.format("Property with name 
'%s' expects type '%s', but got '%s'.", name, unit.setterType, listenerType));
+
+        listeners.put(name, new PropertyChangeListenerWrapper<>(listener, 
listenerType));
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public Set<Entry<String, Object>> entrySet()
+    {
+        return transformSet;
+    }
+
+    private static Map<String, Property> getProperties()
+    {
+        Loader loader = Properties.defaultLoader();
+
+        Map<String, Property> properties = loader.flatten(Config.class);
+        // only handling top-level replacements for now, previous logic was 
only top level so not a regression
+        Map<String, Replacement> replacements = 
Replacements.getNameReplacements(Config.class).get(Config.class);
+        if (replacements != null)
+        {
+            for (Replacement r : replacements.values())
+            {
+                Property latest = properties.get(r.newName);
+                assert latest != null : "Unable to find replacement new name: 
" + r.newName;
+                Property conflict = properties.put(r.oldName, 
r.toProperty(latest));
+                // some configs kept the same name, but changed the type, if 
this is detected then rely on the replaced property
+                assert conflict == null || r.oldName.equals(r.newName) : 
String.format("New property %s attempted to replace %s, but this property 
already exists", latest.getName(), conflict.getName());
+            }
+        }
+
+        for (Map.Entry<String, String> e : 
BACKWARDS_COMPATABLE_NAMES.entrySet())
+        {
+            String oldName = e.getKey();
+            if (properties.containsKey(oldName))
+                throw new AssertionError("Name " + oldName + " is present in 
Config, this adds a conflict as this name had a different meaning.");
+            String newName = e.getValue();
+            Property prop = Objects.requireNonNull(properties.get(newName), 
newName + " cant be found for " + oldName);
+            properties.put(oldName, Properties.rename(oldName, prop));
+        }
+        return properties;
+    }
+
+    /**
+     * settings table was released in 4.0 and attempted to support nested 
properties for a few hand selected properties.
+     * The issue is that 4.0 used '_' to seperate the names, which makes it 
hard to map back to the yaml names; to solve
+     * this 4.1+ uses '.' to avoid possible conflicts, this class provides 
mappings from old names to the '.' names.
+     *
+     * There were a handle full of properties which had custom names, names 
not present in the yaml, this map also
+     * fixes this and returns the proper (what is accessable via yaml) names.
+     */
+    private static Map<String, String> getBackwardsCompatableNames()
+    {
+        Map<String, String> names = new HashMap<>();
+        // Names that don't match yaml
+        names.put("audit_logging_options_logger", 
"audit_logging_options.logger.class_name");
+        names.put("server_encryption_options_client_auth", 
"server_encryption_options.require_client_auth");
+        names.put("server_encryption_options_endpoint_verification", 
"server_encryption_options.require_endpoint_verification");
+        names.put("server_encryption_options_legacy_ssl_storage_port", 
"server_encryption_options.legacy_ssl_storage_port_enabled");
+        names.put("server_encryption_options_protocol", 
"server_encryption_options.accepted_protocols");
+
+        // matching names
+        names.put("audit_logging_options_audit_logs_dir", 
"audit_logging_options.audit_logs_dir");
+        names.put("audit_logging_options_enabled", 
"audit_logging_options.enabled");
+        names.put("audit_logging_options_excluded_categories", 
"audit_logging_options.excluded_categories");
+        names.put("audit_logging_options_excluded_keyspaces", 
"audit_logging_options.excluded_keyspaces");
+        names.put("audit_logging_options_excluded_users", 
"audit_logging_options.excluded_users");
+        names.put("audit_logging_options_included_categories", 
"audit_logging_options.included_categories");
+        names.put("audit_logging_options_included_keyspaces", 
"audit_logging_options.included_keyspaces");
+        names.put("audit_logging_options_included_users", 
"audit_logging_options.included_users");
+        names.put("server_encryption_options_algorithm", 
"server_encryption_options.algorithm");
+        names.put("server_encryption_options_cipher_suites", 
"server_encryption_options.cipher_suites");
+        names.put("server_encryption_options_enabled", 
"server_encryption_options.enabled");
+        names.put("server_encryption_options_internode_encryption", 
"server_encryption_options.internode_encryption");
+        names.put("server_encryption_options_optional", 
"server_encryption_options.optional");
+        names.put("transparent_data_encryption_options_chunk_length_kb", 
"transparent_data_encryption_options.chunk_length_kb");
+        names.put("transparent_data_encryption_options_cipher", 
"transparent_data_encryption_options.cipher");
+        names.put("transparent_data_encryption_options_enabled", 
"transparent_data_encryption_options.enabled");
+        names.put("transparent_data_encryption_options_iv_length", 
"transparent_data_encryption_options.iv_length");
+
+        return names;
+    }
+
+    private static class TransformSet<T, R> extends AbstractSet<R>
+    {
+        private final Set<T> set;
+        private final Function<T, R> transform;
+
+        public TransformSet(Set<T> set, Function<T, R> transform)
+        {
+            this.set = set;
+            this.transform = transform;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public Iterator<R> iterator()
+        {
+            Iterator<T> iter = set.iterator();
+            return new AbstractIterator<R>()
+            {
+                @Override
+                protected R computeNext()
+                {
+                    if (!iter.hasNext())
+                        return endOfData();
+
+                    T entry = iter.next();
+                    return transform.apply(entry);
+                }
+            };
+        }
+        /** {@inheritDoc} */
+        @Override
+        public int size()
+        {
+            return set.size();
+        }
+    }
+
+    private static class PropertyChangeListenerWrapper<T> implements 
PropertyChangeListener<T>
+    {
+        private final PropertyChangeListener<T> listener;
+        private final Class<T> type;
+
+        public PropertyChangeListenerWrapper(PropertyChangeListener<T> 
listener, Class<T> type)
+        {
+            this.listener = listener;
+            this.type = type;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public void onBeforeChange(String name, Object oldValue, Object 
newValue)
+        {
+            T castedOldValue = type.cast(oldValue);
+            T castedNewValue = type.cast(newValue);
+            listener.onBeforeChange(name, castedOldValue, castedNewValue);
+        }
+    }
+
+    /**
+     * Configuration enricher is used to add new property setters to the 
configuration.
+     */
+    private static class ConfigurationSettersEnricher implements 
PropertySetterWalker.SetterVisitor
+    {
+        private final Map<String, ConfigurationUnit<?>> registry;
+
+        ConfigurationSettersEnricher(Map<String, ConfigurationUnit<?>> 
registry)
+        {
+            this.registry = registry;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public <T> void visit(String name, Class<T> clazz, PropertySetter<T> 
setter)
+        {
+            ConfigurationUnit<?> old = registry.get(name);
+            registry.replace(name, new ConfigurationUnit<>(old.source, name, 
old.getter, clazz, setter));
+        }
+    }
+
+    /**
+     * Configuration unit is a single configuration entry that can be read and 
written.
+     * @param <V> type of the configuration setter value.
+     */
+    private static class ConfigurationUnit<V> implements Map.Entry<String, V>

Review Comment:
   im not seeing why we have this class... it looks like it just duplicates 
`Property`. 



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();
+        try
+        {
+            Object oldValue = unit.getValue();
+            if (listeners.get(name) != null)
+                listeners.get(name).onBeforeChange(name, oldValue, value);
+
+            unit.setValue(value);
+            // This potentially may expose the values that are not safe to see 
in logs on production.
+            logger.info("Updated property {} from {} to {}", name, oldValue, 
value);
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Error updating 
property with name '%s', cause: %s", name, e.getMessage()), e);
+        }
+        finally
+        {
+            unit.lock.unlock();
+        }
+    }
+
+    public Class<?> getPropertyType(String key)
+    {
+        if (!registry.containsKey(key))
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", key));
+
+        return registry.get(key).getter.getType();
+    }
+
+    public String getBackwardCompatibleKey(String key)
+    {
+        return BACKWARDS_COMPATABLE_NAMES.get(key);
+    }
+
+    /**
+     * Adds a listener for the property with the given name.
+     * @param name property name to listen to.
+     * @param listener listener to add.
+     * @param <T> type of the property.
+     */
+    public <T> void addPropertyChangeListener(String name, 
PropertyChangeListener<T> listener, Class<T> listenerType)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe for resitering handler.", name));
+
+        if (unit.setterType == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+//        Class<T> type = (Class<T>) ((ParameterizedType) 
listener.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
+        if (!unit.setterType.equals(listenerType))
+            throw new ConfigurationException(String.format("Property with name 
'%s' expects type '%s', but got '%s'.", name, unit.setterType, listenerType));
+
+        listeners.put(name, new PropertyChangeListenerWrapper<>(listener, 
listenerType));
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public Set<Entry<String, Object>> entrySet()
+    {
+        return transformSet;
+    }
+
+    private static Map<String, Property> getProperties()
+    {
+        Loader loader = Properties.defaultLoader();
+
+        Map<String, Property> properties = loader.flatten(Config.class);
+        // only handling top-level replacements for now, previous logic was 
only top level so not a regression
+        Map<String, Replacement> replacements = 
Replacements.getNameReplacements(Config.class).get(Config.class);
+        if (replacements != null)
+        {
+            for (Replacement r : replacements.values())
+            {
+                Property latest = properties.get(r.newName);
+                assert latest != null : "Unable to find replacement new name: 
" + r.newName;
+                Property conflict = properties.put(r.oldName, 
r.toProperty(latest));
+                // some configs kept the same name, but changed the type, if 
this is detected then rely on the replaced property
+                assert conflict == null || r.oldName.equals(r.newName) : 
String.format("New property %s attempted to replace %s, but this property 
already exists", latest.getName(), conflict.getName());
+            }
+        }
+
+        for (Map.Entry<String, String> e : 
BACKWARDS_COMPATABLE_NAMES.entrySet())
+        {
+            String oldName = e.getKey();
+            if (properties.containsKey(oldName))
+                throw new AssertionError("Name " + oldName + " is present in 
Config, this adds a conflict as this name had a different meaning.");
+            String newName = e.getValue();
+            Property prop = Objects.requireNonNull(properties.get(newName), 
newName + " cant be found for " + oldName);
+            properties.put(oldName, Properties.rename(oldName, prop));
+        }
+        return properties;
+    }
+
+    /**
+     * settings table was released in 4.0 and attempted to support nested 
properties for a few hand selected properties.
+     * The issue is that 4.0 used '_' to seperate the names, which makes it 
hard to map back to the yaml names; to solve
+     * this 4.1+ uses '.' to avoid possible conflicts, this class provides 
mappings from old names to the '.' names.
+     *
+     * There were a handle full of properties which had custom names, names 
not present in the yaml, this map also
+     * fixes this and returns the proper (what is accessable via yaml) names.
+     */
+    private static Map<String, String> getBackwardsCompatableNames()

Review Comment:
   should not rely on copy/paste, the 2 may drift and will be hard to maintain



##########
src/java/org/apache/cassandra/config/ConfigFields.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config;
+
+/**
+ * This class is autogenerated bases on the {@link Config} class. It contains 
all the configuration property

Review Comment:
   > This class is autogenerated
   
   Where?  I don't see anything in this patch that generates it, `build.xml` 
wasn't updated and this is not in a `gen-java` dir



##########
src/java/org/apache/cassandra/config/DatabaseDescriptorWalker.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config;
+
+import org.apache.cassandra.config.registry.PropertySetterWalker;
+
+/**
+ * This class is autogenerated bases on given {@link DatabaseDescriptor} 
class. It contains all the configuration

Review Comment:
   > This class is autogenerated
   
   Where?  I don't see such a thing in this patch



##########
src/java/org/apache/cassandra/db/virtual/SettingsTable.java:
##########
@@ -59,102 +56,83 @@ final class SettingsTable extends AbstractVirtualTable
                            .addPartitionKeyColumn(NAME, UTF8Type.instance)
                            .addRegularColumn(VALUE, UTF8Type.instance)
                            .build());
-        this.config = config;
+        registerConverters();
+        this.configurationRegistry = configurationRegistry;
     }
 
-    @Override
-    public DataSet data(DecoratedKey partitionKey)
+    private void registerConverters()
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        String name = UTF8Type.instance.compose(partitionKey.getKey());
-        if (BACKWARDS_COMPATABLE_NAMES.containsKey(name))
-            ClientWarn.instance.warn("key '" + name + "' is deprecated; should 
switch to '" + BACKWARDS_COMPATABLE_NAMES.get(name) + "'");
-        if (PROPERTIES.containsKey(name))
-            result.row(name).column(VALUE, getValue(PROPERTIES.get(name)));
-        return result;
+        converterRegistry.put(Boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(Integer.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(int.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(DurationSpec.LongMillisecondsBound.class,
+                              
(PropertyConverter<DurationSpec.LongMillisecondsBound>) value -> 
DurationSpec.from(value, DurationSpec.LongMillisecondsBound.class));
     }
 
-    @Override
-    public DataSet data()
+    /**
+     * Setter for the property.
+     * @param name the name of the property.
+     * @param value the string representation of the value of the property to 
set.
+     */
+    private void setProperty(String name, String value)
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        for (Map.Entry<String, Property> e : PROPERTIES.entrySet())
-            result.row(e.getKey()).column(VALUE, getValue(e.getValue()));
-        return result;
+        Class<?> setterType = configurationRegistry.getPropertyType(name);
+        PropertyConverter<?> converter = converterRegistry.get(setterType);
+        if (converter == null)
+            throw invalidRequest("Unknown converter for property with name 
'%s' and type '%s'", name, setterType);
+
+        configurationRegistry.update(name, value == null ? null : 
converter.convert(value));
     }
 
-    private String getValue(Property prop)
+    private static @Nullable String getProperty(ConfigurationRegistry 
registry, String name)
     {
-        Object value = prop.get(config);
+        Object value = registry.get(name);
         return value == null ? null : value.toString();

Review Comment:
   an issue this patch doesn't look to address is that the `input` and `output` 
may not match.
   
   For basic types like int/boolean this is fine as they are simple, but cases 
like `hinted_handoff_disabled_datacenters` are not correct as that's java 
collection toString, which we won't have a property converter for...  it's also 
weird for users to generate java's collection toString.  There are many types 
that fall into this bucket.
   
   You *should* have a property that w/e is returned from this table is 
accepted by this table; which is not true with this patch.



##########
src/java/org/apache/cassandra/db/virtual/SettingsTable.java:
##########
@@ -59,102 +56,83 @@ final class SettingsTable extends AbstractVirtualTable
                            .addPartitionKeyColumn(NAME, UTF8Type.instance)
                            .addRegularColumn(VALUE, UTF8Type.instance)
                            .build());
-        this.config = config;
+        registerConverters();
+        this.configurationRegistry = configurationRegistry;
     }
 
-    @Override
-    public DataSet data(DecoratedKey partitionKey)
+    private void registerConverters()
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        String name = UTF8Type.instance.compose(partitionKey.getKey());
-        if (BACKWARDS_COMPATABLE_NAMES.containsKey(name))
-            ClientWarn.instance.warn("key '" + name + "' is deprecated; should 
switch to '" + BACKWARDS_COMPATABLE_NAMES.get(name) + "'");
-        if (PROPERTIES.containsKey(name))
-            result.row(name).column(VALUE, getValue(PROPERTIES.get(name)));
-        return result;
+        converterRegistry.put(Boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(Integer.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(int.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(DurationSpec.LongMillisecondsBound.class,

Review Comment:
   why are you using reflection when you can do 
`DurationSpec.LongMillisecondsBound::new`?



##########
src/java/org/apache/cassandra/db/virtual/SettingsTable.java:
##########
@@ -59,102 +56,83 @@ final class SettingsTable extends AbstractVirtualTable
                            .addPartitionKeyColumn(NAME, UTF8Type.instance)
                            .addRegularColumn(VALUE, UTF8Type.instance)
                            .build());
-        this.config = config;
+        registerConverters();
+        this.configurationRegistry = configurationRegistry;
     }
 
-    @Override
-    public DataSet data(DecoratedKey partitionKey)
+    private void registerConverters()
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        String name = UTF8Type.instance.compose(partitionKey.getKey());
-        if (BACKWARDS_COMPATABLE_NAMES.containsKey(name))
-            ClientWarn.instance.warn("key '" + name + "' is deprecated; should 
switch to '" + BACKWARDS_COMPATABLE_NAMES.get(name) + "'");
-        if (PROPERTIES.containsKey(name))
-            result.row(name).column(VALUE, getValue(PROPERTIES.get(name)));
-        return result;
+        converterRegistry.put(Boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(boolean.class, 
CassandraRelevantProperties.BOOLEAN_CONVERTER);
+        converterRegistry.put(Integer.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(int.class, 
CassandraRelevantProperties.INTEGER_CONVERTER);
+        converterRegistry.put(DurationSpec.LongMillisecondsBound.class,
+                              
(PropertyConverter<DurationSpec.LongMillisecondsBound>) value -> 
DurationSpec.from(value, DurationSpec.LongMillisecondsBound.class));
     }
 
-    @Override
-    public DataSet data()
+    /**
+     * Setter for the property.
+     * @param name the name of the property.
+     * @param value the string representation of the value of the property to 
set.
+     */
+    private void setProperty(String name, String value)
     {
-        SimpleDataSet result = new SimpleDataSet(metadata());
-        for (Map.Entry<String, Property> e : PROPERTIES.entrySet())
-            result.row(e.getKey()).column(VALUE, getValue(e.getValue()));
-        return result;
+        Class<?> setterType = configurationRegistry.getPropertyType(name);
+        PropertyConverter<?> converter = converterRegistry.get(setterType);
+        if (converter == null)
+            throw invalidRequest("Unknown converter for property with name 
'%s' and type '%s'", name, setterType);
+
+        configurationRegistry.update(name, value == null ? null : 
converter.convert(value));

Review Comment:
   I don't follow why this is needed.  You have a map of "real" type to 
"PropertyConverter", so can't you just do
   
   ```
   Property p = PROPERTIES.get(name);
   if (p == null) throw unknownProperty(name);
   p.set(DatabaseDescriptor.getRawConfig(), 
getConverter(p.getType()).convert(value));
   ```
   
   If there is a missing converter that means we have a bug, we don't support 
all types we support...



##########
src/java/org/apache/cassandra/config/PropertyConverter.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config;
+
+/**
+ * A converter that converts a string to a specific type.
+ * @param <T> the type to convert to
+ */
+public interface PropertyConverter<T>

Review Comment:
   you moved this out of the `CassandraRelevantProperties`, so are you making a 
assertion that system properties *must* match the same input as the vtable?  I 
am not saying they shouldn't, I am calling out that is what this code is 
asserting.  The current usage are all primitives, so this is fine, but we 
should call out that we don't actual handle `ConfigurationException` in the 
vtable, so user will get an error saying C* failed, and not that we had bad 
input



##########
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java:
##########
@@ -119,6 +120,16 @@ public void test() throws Exception
                 throw root;
             }
         }
+
+        long ts = currentTimeMillis();
+        while (!Thread.interrupted())
+        {
+            if (ts + 5000 >= currentTimeMillis())
+                continue;
+
+            ts = currentTimeMillis();
+            System.out.println(">>>> Running: " + ts);
+        }

Review Comment:
   why is this here?  Also should avoid `System.out`



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());

Review Comment:
   I don't understand the need for `PropertySetterWalker` and this field?  
There exists a `Property` for every key (nested and top level), so I think we 
could remove a lot of code by just calling `Property.set` like we do in YAML



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();
+        try
+        {
+            Object oldValue = unit.getValue();
+            if (listeners.get(name) != null)
+                listeners.get(name).onBeforeChange(name, oldValue, value);
+
+            unit.setValue(value);
+            // This potentially may expose the values that are not safe to see 
in logs on production.
+            logger.info("Updated property {} from {} to {}", name, oldValue, 
value);
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Error updating 
property with name '%s', cause: %s", name, e.getMessage()), e);
+        }
+        finally
+        {
+            unit.lock.unlock();
+        }
+    }
+
+    public Class<?> getPropertyType(String key)
+    {
+        if (!registry.containsKey(key))
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", key));
+
+        return registry.get(key).getter.getType();
+    }
+
+    public String getBackwardCompatibleKey(String key)
+    {
+        return BACKWARDS_COMPATABLE_NAMES.get(key);
+    }
+
+    /**
+     * Adds a listener for the property with the given name.
+     * @param name property name to listen to.
+     * @param listener listener to add.
+     * @param <T> type of the property.
+     */
+    public <T> void addPropertyChangeListener(String name, 
PropertyChangeListener<T> listener, Class<T> listenerType)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe for resitering handler.", name));
+
+        if (unit.setterType == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+//        Class<T> type = (Class<T>) ((ParameterizedType) 
listener.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
+        if (!unit.setterType.equals(listenerType))
+            throw new ConfigurationException(String.format("Property with name 
'%s' expects type '%s', but got '%s'.", name, unit.setterType, listenerType));
+
+        listeners.put(name, new PropertyChangeListenerWrapper<>(listener, 
listenerType));

Review Comment:
   what happens if multiple different places need to listen to the same config 
key?



##########
src/java/org/apache/cassandra/config/registry/ConfigurationRegistry.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config.registry;
+
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.introspector.Property;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DatabaseDescriptorWalker;
+import org.apache.cassandra.config.Loader;
+import org.apache.cassandra.config.Properties;
+import org.apache.cassandra.config.Replacement;
+import org.apache.cassandra.config.Replacements;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.AbstractIterator;
+
+
+/**
+ * This is a simple configuration property registry that stores all the {@link 
Config} settings.
+ * It is a singleton and can be accessed via {@link #instance} field.
+ */
+public class ConfigurationRegistry extends AbstractMap<String, Object>
+{
+    public static final List<PropertySetterWalker> PROPERTY_SETTERS_LIST = 
ImmutableList.of(new DatabaseDescriptorWalker());
+    private static final Map<String, String> BACKWARDS_COMPATABLE_NAMES = 
ImmutableMap.copyOf(getBackwardsCompatableNames());
+    private static final Logger logger = 
LoggerFactory.getLogger(ConfigurationRegistry.class);
+    public static final ConfigurationRegistry instance = new 
ConfigurationRegistry();
+    private final Map<String, ConfigurationUnit<?>> registry;
+    private final TransformSet<Map.Entry<String, ConfigurationUnit<?>>, 
Map.Entry<String, Object>> transformSet;
+    private final Map<String, PropertyChangeListenerWrapper<?>> listeners = 
new HashMap<>();
+
+    public ConfigurationRegistry()
+    {
+        this(DatabaseDescriptor.getRawConfig());
+    }
+
+    public ConfigurationRegistry(Config config)
+    {
+        registry = getProperties().entrySet()
+                                  .stream()
+                                  .collect(Collectors.toMap(Map.Entry::getKey, 
e -> new ConfigurationUnit<>(config, e.getKey(), e.getValue())));
+
+        ConfigurationSettersEnricher enricher = new 
ConfigurationSettersEnricher(registry);
+        PROPERTY_SETTERS_LIST.forEach(w -> w.walk(enricher));
+        transformSet = new TransformSet<>(registry.entrySet(),
+                                          e -> new 
AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getValue()));
+    }
+
+    /**
+     * Setter for the property with the given name. Can accept {@code null} 
value.
+     * @param name the name of the property.
+     * @param value the value to set.
+     */
+    public void update(String name, Object value)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", name));
+
+        if (unit.setter == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+        unit.lock.lock();
+        try
+        {
+            Object oldValue = unit.getValue();
+            if (listeners.get(name) != null)
+                listeners.get(name).onBeforeChange(name, oldValue, value);
+
+            unit.setValue(value);
+            // This potentially may expose the values that are not safe to see 
in logs on production.
+            logger.info("Updated property {} from {} to {}", name, oldValue, 
value);
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Error updating 
property with name '%s', cause: %s", name, e.getMessage()), e);
+        }
+        finally
+        {
+            unit.lock.unlock();
+        }
+    }
+
+    public Class<?> getPropertyType(String key)
+    {
+        if (!registry.containsKey(key))
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe.", key));
+
+        return registry.get(key).getter.getType();
+    }
+
+    public String getBackwardCompatibleKey(String key)
+    {
+        return BACKWARDS_COMPATABLE_NAMES.get(key);
+    }
+
+    /**
+     * Adds a listener for the property with the given name.
+     * @param name property name to listen to.
+     * @param listener listener to add.
+     * @param <T> type of the property.
+     */
+    public <T> void addPropertyChangeListener(String name, 
PropertyChangeListener<T> listener, Class<T> listenerType)
+    {
+        ConfigurationUnit<?> unit = registry.get(name);
+        if (unit == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not availabe for resitering handler.", name));
+
+        if (unit.setterType == null)
+            throw new ConfigurationException(String.format("Property with name 
'%s' is not writable.", name));
+
+//        Class<T> type = (Class<T>) ((ParameterizedType) 
listener.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0];
+        if (!unit.setterType.equals(listenerType))
+            throw new ConfigurationException(String.format("Property with name 
'%s' expects type '%s', but got '%s'.", name, unit.setterType, listenerType));
+
+        listeners.put(name, new PropertyChangeListenerWrapper<>(listener, 
listenerType));
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public Set<Entry<String, Object>> entrySet()

Review Comment:
   this is very confusing, this boils down to the state in `registry`, but if I 
do `cr.put("does not exist", null)` I won't see that if I call `entrySet`



##########
src/java/org/apache/cassandra/config/ConfigFields.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.config;
+
+/**
+ * This class is autogenerated bases on the {@link Config} class. It contains 
all the configuration property
+ * names as constants.
+ */
+public class ConfigFields

Review Comment:
   I don't follow this logic... why is there another config system rather than 
just updating `Config`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org
For additional commands, e-mail: pr-h...@cassandra.apache.org