diff --git a/config/log4j2.properties b/config/log4j2.properties
index 1c676fefecb..92acf87be41 100644
--- a/config/log4j2.properties
+++ b/config/log4j2.properties
@@ -154,7 +154,7 @@ appender.deprecation_rolling.policies.size.size = 100MB
 appender.deprecation_rolling.strategy.type = DefaultRolloverStrategy
 appender.deprecation_rolling.strategy.max = 30
 
-logger.deprecation.name = org.logstash.deprecation, deprecation
+logger.deprecation.name = org.logstash.deprecation
 logger.deprecation.level = WARN
 logger.deprecation.appenderRef.deprecation_rolling.ref = deprecation_plain_rolling
 logger.deprecation.additivity = false
diff --git a/logstash-core/lib/logstash/settings.rb b/logstash-core/lib/logstash/settings.rb
index d8c75f220af..d5381c9779f 100644
--- a/logstash-core/lib/logstash/settings.rb
+++ b/logstash-core/lib/logstash/settings.rb
@@ -86,7 +86,10 @@ def initialize
     end
 
     def register(setting)
-      return setting.map { |s| register(s) } if setting.kind_of?(Array)
+      # Method #with_deprecated_alias returns collection containing couple of other settings.
+      # In case the method is implemented in Ruby returns an Array while for the Java implementation
+      # return a List, so the following type checking before going deep by one layer.
+      return setting.map { |s| register(s) } if setting.kind_of?(Array) || setting.kind_of?(java.util.List)
 
       if @settings.key?(setting.name)
         raise ArgumentError.new("Setting \"#{setting.name}\" has already been registered as #{setting.inspect}")
@@ -244,54 +247,73 @@ def flatten_hash(h, f = "", g = {})
   class Setting
     include LogStash::Settings::LOGGABLE_PROXY
 
-    attr_reader :name, :default
+    attr_reader :wrapped_setting
 
     def initialize(name, klass, default = nil, strict = true, &validator_proc)
-      @name = name
       unless klass.is_a?(Class)
-        raise ArgumentError.new("Setting \"#{@name}\" must be initialized with a class (received #{klass})")
+        raise ArgumentError.new("Setting \"#{name}\" must be initialized with a class (received #{klass})")
       end
+      setting_builder = Java::org.logstash.settings.BaseSetting.create(name)
+                            .defaultValue(default)
+                            .strict(strict)
+      if validator_proc
+        setting_builder = setting_builder.validator(validator_proc)
+      end
+
+      @wrapped_setting = setting_builder.build()
+
       @klass = klass
       @validator_proc = validator_proc
-      @value = nil
-      @value_is_set = false
-      @strict = strict
 
-      validate(default) if @strict
-      @default = default
+      validate(default) if strict?
+    end
+
+    def default
+      @wrapped_setting.default
+    end
+
+    def name
+      @wrapped_setting.name
+    end
+
+    def initialize_copy(original)
+      @wrapped_setting = original.wrapped_setting.clone
+    end
+
+    # To be used only internally
+    def update_wrapper(wrapped_setting)
+      @wrapped_setting = wrapped_setting
     end
 
     def value
-      @value_is_set ? @value : default
+      @wrapped_setting.value()
     end
 
     def set?
-      @value_is_set
+      @wrapped_setting.set?
     end
 
     def strict?
-      @strict
+      @wrapped_setting.strict?
     end
 
     def set(value)
-      validate(value) if @strict
-      @value = value
-      @value_is_set = true
-      @value
+      validate(value) if strict?
+      @wrapped_setting.set(value)
+      @wrapped_setting.value
     end
 
     def reset
-      @value = nil
-      @value_is_set = false
+      @wrapped_setting.reset
     end
 
     def to_hash
       {
-        "name" => @name,
+        "name" => @wrapped_setting.name,
         "klass" => @klass,
-        "value" => @value,
-        "value_is_set" => @value_is_set,
-        "default" => @default,
+        "value" => @wrapped_setting.value,
+        "value_is_set" => @wrapped_setting.set?,
+        "default" => @wrapped_setting.default,
         # Proc#== will only return true if it's the same obj
         # so no there's no point in comparing it
         # also thereś no use case atm to return the proc
@@ -301,7 +323,7 @@ def to_hash
     end
 
     def inspect
-      "<#{self.class.name}(#{name}): #{value.inspect}" + (@value_is_set ? '' : ' (DEFAULT)') + ">"
+      "<#{self.class.name}(#{name}): #{value.inspect}" + (@wrapped_setting.set? ? '' : ' (DEFAULT)') + ">"
     end
 
     def ==(other)
@@ -323,58 +345,65 @@ def nullable
     end
 
     def format(output)
-      effective_value = self.value
-      default_value = self.default
-      setting_name = self.name
+      @wrapped_setting.format(output)
+    end
 
-      if default_value == value # print setting and its default value
-        output << "#{setting_name}: #{effective_value.inspect}" unless effective_value.nil?
-      elsif default_value.nil? # print setting and warn it has been set
-        output << "*#{setting_name}: #{effective_value.inspect}"
-      elsif effective_value.nil? # default setting not set by user
-        output << "#{setting_name}: #{default_value.inspect}"
-      else # print setting, warn it has been set, and show default value
-        output << "*#{setting_name}: #{effective_value.inspect} (default: #{default_value.inspect})"
-      end
+    def clone(*args)
+      copy = self.dup
+      copy.update_wrapper(@wrapped_setting.clone())
+      copy
     end
 
     protected
     def validate(input)
       if !input.is_a?(@klass)
-        raise ArgumentError.new("Setting \"#{@name}\" must be a #{@klass}. Received: #{input} (#{input.class})")
+        raise ArgumentError.new("Setting \"#{@wrapped_setting.name}\" must be a #{@klass}. Received: #{input} (#{input.class})")
       end
 
       if @validator_proc && !@validator_proc.call(input)
-        raise ArgumentError.new("Failed to validate setting \"#{@name}\" with value: #{input}")
+        raise ArgumentError.new("Failed to validate setting \"#{@wrapped_setting.name}\" with value: #{input}")
       end
     end
 
     class Coercible < Setting
       def initialize(name, klass, default = nil, strict = true, &validator_proc)
-        @name = name
         unless klass.is_a?(Class)
-          raise ArgumentError.new("Setting \"#{@name}\" must be initialized with a class (received #{klass})")
+          raise ArgumentError.new("Setting \"#{name}\" must be initialized with a class (received #{klass})")
         end
+
         @klass = klass
         @validator_proc = validator_proc
-        @value = nil
-        @value_is_set = false
+
+        # needed to have the name method accessible when invoking validate
+        @wrapped_setting = Java::org.logstash.settings.BaseSetting.create(name)
+                                      .defaultValue(default)
+                                      .strict(strict)
+                                      .build()
 
         if strict
           coerced_default = coerce(default)
           validate(coerced_default)
-          @default = coerced_default
+          updated_default = coerced_default
         else
-          @default = default
+          updated_default = default
+        end
+
+        # default value must be coerced to the right type before being set
+        setting_builder = Java::org.logstash.settings.BaseSetting.create(name)
+                              .defaultValue(updated_default)
+                              .strict(strict)
+        if validator_proc
+          setting_builder = setting_builder.validator(validator_proc)
         end
+
+        @wrapped_setting = setting_builder.build()
       end
 
       def set(value)
         coerced_value = coerce(value)
         validate(coerced_value)
-        @value = coerce(coerced_value)
-        @value_is_set = true
-        @value
+        @wrapped_setting.set(coerced_value)
+        coerced_value
       end
 
       def coerce(value)
@@ -383,22 +412,7 @@ def coerce(value)
     end
     ### Specific settings #####
 
-    class Boolean < Coercible
-      def initialize(name, default, strict = true, &validator_proc)
-        super(name, Object, default, strict, &validator_proc)
-      end
-
-      def coerce(value)
-        case value
-        when TrueClass, "true"
-          true
-        when FalseClass, "false"
-          false
-        else
-          raise ArgumentError.new("could not coerce #{value} into a boolean")
-        end
-      end
-    end
+    java_import org.logstash.settings.Boolean
 
     class Numeric < Coercible
       def initialize(name, default = nil, strict = true)
@@ -733,15 +747,15 @@ def coerce(value)
       protected
       def validate(input)
         if !input.is_a?(@klass)
-          raise ArgumentError.new("Setting \"#{@name}\" must be a #{@klass}. Received: #{input} (#{input.class})")
+          raise ArgumentError.new("Setting \"#{@wrapped_setting.name}\" must be a #{@klass}. Received: #{input} (#{input.class})")
         end
 
         unless input.all? {|el| el.kind_of?(@element_class) }
-          raise ArgumentError.new("Values of setting \"#{@name}\" must be #{@element_class}. Received: #{input.map(&:class)}")
+          raise ArgumentError.new("Values of setting \"#{@wrapped_setting.name}\" must be #{@element_class}. Received: #{input.map(&:class)}")
         end
 
         if @validator_proc && !@validator_proc.call(input)
-          raise ArgumentError.new("Failed to validate setting \"#{@name}\" with value: #{input}")
+          raise ArgumentError.new("Failed to validate setting \"#{@wrapped_setting.name}\" with value: #{input}")
         end
       end
     end
@@ -782,7 +796,7 @@ def validate(value)
         return unless invalid_value.any?
 
         raise ArgumentError,
-          "Failed to validate the setting \"#{@name}\" value(s): #{invalid_value.inspect}. Valid options are: #{@possible_strings.inspect}"
+          "Failed to validate the setting \"#{@wrapped_setting.name}\" value(s): #{invalid_value.inspect}. Valid options are: #{@possible_strings.inspect}"
       end
     end
 
@@ -792,9 +806,9 @@ def initialize(name, klass, default = nil)
       end
 
       def set(value)
-        @value = coerce(value)
-        @value_is_set = true
-        @value
+        coerced_value = coerce(value)
+        @wrapped_setting.set(coerced_value)
+        coerced_value
       end
 
       def coerce(value)
@@ -839,8 +853,7 @@ def initialize(canonical_proxy, alias_name)
         @canonical_proxy = canonical_proxy
 
         clone = @canonical_proxy.canonical_setting.clone
-        clone.instance_variable_set(:@name, alias_name)
-        clone.instance_variable_set(:@default, nil)
+        clone.update_wrapper(clone.wrapped_setting.deprecate(alias_name))
 
         super(clone)
       end
diff --git a/logstash-core/src/main/java/org/logstash/settings/BaseSetting.java b/logstash-core/src/main/java/org/logstash/settings/BaseSetting.java
new file mode 100644
index 00000000000..b08408393df
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/BaseSetting.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+/**
+ * Root class for all setting definitions.
+ * */
+public class BaseSetting<T> implements Setting<T> {
+
+    private String name; // not final because can be updated by deprecate
+    T defaultValue;
+    private T value = null;
+    private final boolean strict;
+    private final Predicate<T> validator;
+    private boolean valueIsSet = false;
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public BaseSetting<T> clone() {
+        try {
+            BaseSetting<T> clone = (BaseSetting<T>) super.clone();
+            // copy mutable state here, so the clone can't change the internals of the original
+            clone.value = value;
+            clone.valueIsSet = valueIsSet;
+            return clone;
+        } catch (CloneNotSupportedException e) {
+            throw new AssertionError();
+        }
+    }
+
+    public static final class Builder<T> {
+        private final String name;
+        private boolean strict = true;
+        private T defaultValue = null;
+        private Predicate<T> validator = noValidator();
+
+        public Builder(String name) {
+            this.name = name;
+        }
+
+        public Builder<T> defaultValue(T defaultValue) {
+            this.defaultValue = defaultValue;
+            return this;
+        }
+
+        public Builder<T> strict(boolean strict) {
+            this.strict = strict;
+            return this;
+        }
+
+        public Builder<T> validator(Predicate<T> validator) {
+            this.validator = validator;
+            return this;
+        }
+
+        public BaseSetting<T> build() {
+            return new BaseSetting<>(name, defaultValue, strict, validator);
+        }
+    }
+
+    public static <T> Builder<T> create(String name) {
+        return new Builder<>(name);
+    }
+
+    /**
+     * Specifically used by Coercible subclass to initialize doing validation in a second phase.
+     * */
+    protected BaseSetting(String name, boolean strict, Predicate<T> validator) {
+        Objects.requireNonNull(name);
+        Objects.requireNonNull(validator);
+        this.name = name;
+        this.strict = strict;
+        this.validator = validator;
+    }
+
+    @SuppressWarnings("this-escape")
+    protected BaseSetting(String name, T defaultValue, boolean strict, Predicate<T> validator) {
+        Objects.requireNonNull(name);
+        Objects.requireNonNull(validator);
+        this.name = name;
+        this.defaultValue = defaultValue;
+        this.strict = strict;
+        this.validator = validator;
+        if (strict) {
+           validate(defaultValue);
+        }
+    }
+
+    /**
+     * Creates a copy of the setting with the original name to deprecate
+     * */
+    protected BaseSetting<T> deprecate(String deprecatedName) {
+        // this force to get a copy of the original Setting, in case of a BooleanSetting it retains also all of its
+        // coercing mechanisms
+        BaseSetting<T> clone = this.clone();
+        clone.updateName(deprecatedName);
+        return clone;
+    }
+
+    private void updateName(String deprecatedName) {
+        this.name = deprecatedName;
+    }
+
+    protected static <T> Predicate<T> noValidator() {
+        return t -> true;
+    }
+
+    public void validate(T input) throws IllegalArgumentException {
+        if (!validator.test(input)) {
+            throw new IllegalArgumentException("Failed to validate setting " + this.name + " with value: " + input);
+        }
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public T value() {
+        if (valueIsSet) {
+            return value;
+        } else {
+            return defaultValue;
+        }
+    }
+
+    public boolean isSet() {
+        return this.valueIsSet;
+    }
+
+    public boolean isStrict() {
+        return strict;
+    }
+
+    public void setSafely(T newValue) {
+        if (strict) {
+            validate(newValue);
+        }
+        this.value = newValue;
+        this.valueIsSet = true;
+    }
+
+    public void reset() {
+        this.value = null;
+        this.valueIsSet = false;
+    }
+
+    public void validateValue() {
+        validate(this.value);
+    }
+
+    public T getDefault() {
+        return this.defaultValue;
+    }
+
+    public void format(List<String> output) {
+        T effectiveValue = this.value;
+        String settingName = this.name;
+
+        if (effectiveValue != null && effectiveValue.equals(defaultValue)) {
+            // print setting and its default value
+            output.add(String.format("%s: %s", settingName, effectiveValue));
+        } else if (defaultValue == null) {
+            // print setting and warn it has been set
+            output.add(String.format("*%s: %s", settingName, effectiveValue));
+        } else if (effectiveValue == null) {
+            // default setting not set by user
+            output.add(String.format("%s: %s", settingName, defaultValue));
+        } else {
+            // print setting, warn it has been set, and show default value
+            output.add(String.format("*%s: %s (default: %s)", settingName, effectiveValue, defaultValue));
+        }
+    }
+
+    public List<Setting<T>> withDeprecatedAlias(String deprecatedAlias) {
+        return SettingWithDeprecatedAlias.wrap(this, deprecatedAlias);
+    }
+
+    public Setting<T> nullable() {
+        return new NullableSetting<>(this);
+    }
+ }
+
+
+
diff --git a/logstash-core/src/main/java/org/logstash/settings/Boolean.java b/logstash-core/src/main/java/org/logstash/settings/Boolean.java
new file mode 100644
index 00000000000..2b9feb6f485
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/Boolean.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+public class Boolean extends Coercible<java.lang.Boolean> {
+
+    public Boolean(String name, boolean defaultValue) {
+        super(name, defaultValue, true, noValidator());
+    }
+
+    public Boolean(String name, boolean defaultValue, boolean strict) {
+        super(name, defaultValue, strict, noValidator());
+    }
+
+    @Override
+    public java.lang.Boolean coerce(Object obj) {
+        if (obj instanceof String) {
+            switch((String) obj) {
+                case "true": return true;
+                case "false": return false;
+                default: throw new IllegalArgumentException(coercionFailureMessage(obj));
+            }
+        }
+        if (obj instanceof java.lang.Boolean) {
+            return (java.lang.Boolean) obj;
+        }
+        throw new IllegalArgumentException(coercionFailureMessage(obj));
+    }
+
+    private String coercionFailureMessage(Object obj) {
+        return String.format("Cannot coerce `%s` to boolean (%s)", obj, getName());
+    }
+}
\ No newline at end of file
diff --git a/logstash-core/src/main/java/org/logstash/settings/Coercible.java b/logstash-core/src/main/java/org/logstash/settings/Coercible.java
new file mode 100644
index 00000000000..ec6688969d5
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/Coercible.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import java.util.function.Predicate;
+
+public abstract class Coercible<T> extends BaseSetting<T> {
+
+    @SuppressWarnings("this-escape")
+    public Coercible(String name, T defaultValue, boolean strict, Predicate<T> validator) {
+        super(name, strict, validator);
+
+        if (strict) {
+            T coercedDefault = coerce(defaultValue);
+            validate(coercedDefault);
+            this.defaultValue = coercedDefault;
+        } else {
+            this.defaultValue = defaultValue;
+        }
+    }
+
+    @Override
+    public void set(Object value) {
+        T coercedValue = coerce(value);
+        validate(coercedValue);
+        super.setSafely(coercedValue);
+    }
+
+    @Override
+    public void setSafely(T value) {
+        this.set(value);
+    }
+
+    public abstract T coerce(Object obj);
+}
diff --git a/logstash-core/src/main/java/org/logstash/settings/DeprecatedAlias.java b/logstash-core/src/main/java/org/logstash/settings/DeprecatedAlias.java
new file mode 100644
index 00000000000..d2b8eac17e3
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/DeprecatedAlias.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import co.elastic.logstash.api.DeprecationLogger;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.logstash.log.DefaultDeprecationLogger;
+
+/**
+ * A <code>DeprecatedAlias</code> provides a deprecated alias for a setting, and is meant
+ * to be used exclusively through @see org.logstash.settings.SettingWithDeprecatedAlias#wrap()
+ * */
+public final class DeprecatedAlias<T> extends SettingDelegator<T> {
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    private static final DeprecationLogger DEPRECATION_LOGGER = new DefaultDeprecationLogger(LOGGER);
+
+    private SettingWithDeprecatedAlias<T> canonicalProxy;
+
+    DeprecatedAlias(SettingWithDeprecatedAlias<T> canonicalProxy, String aliasName) {
+        super(canonicalProxy.getCanonicalSetting().deprecate(aliasName));
+        this.canonicalProxy = canonicalProxy;
+    }
+
+    // Because loggers are configure after the Settings declaration, this method is intended for lazy-logging
+    // check https://github.com/elastic/logstash/pull/16339
+    public void observePostProcess() {
+        if (isSet()) {
+            DEPRECATION_LOGGER.deprecated("The setting `{}` is a deprecated alias for `{}` and will be removed in a " +
+                    "future release of Logstash. Please use `{}` instead", getName(), canonicalProxy.getName(), canonicalProxy.getName());
+        }
+    }
+
+    @Override
+    public T value() {
+        LOGGER.warn("The value of setting `{}` has been queried by its deprecated alias `{}`. " +
+                "Code should be updated to query `{}` instead", canonicalProxy.getName(), getName(), canonicalProxy.getName());
+        return super.value();
+    }
+
+    @Override
+    public void validateValue() {
+        // bypass deprecation warning
+        if (isSet()) {
+            getDelegate().validateValue();
+        }
+    }
+}
diff --git a/logstash-core/src/main/java/org/logstash/settings/NullableSetting.java b/logstash-core/src/main/java/org/logstash/settings/NullableSetting.java
new file mode 100644
index 00000000000..0b1b27e3cfb
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/NullableSetting.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+public class NullableSetting<T> extends SettingDelegator<T> {
+
+    NullableSetting(BaseSetting<T> delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public void validate(T input) throws IllegalArgumentException {
+        if (input == null) {
+            return;
+        }
+        getDelegate().validate(input);
+    }
+
+    // prevent delegate from intercepting
+    @Override
+    public void validateValue() {
+        validate(value());
+    }
+}
diff --git a/logstash-core/src/main/java/org/logstash/settings/Setting.java b/logstash-core/src/main/java/org/logstash/settings/Setting.java
new file mode 100644
index 00000000000..f97f9055c54
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/Setting.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import java.util.List;
+
+public interface Setting<T> extends Cloneable {
+
+    String getName();
+
+    T value();
+
+    boolean isSet();
+
+    boolean isStrict();
+
+    void setSafely(T newValue);
+
+    @SuppressWarnings("unchecked")
+    default void set(Object newValue) {
+        //this could throw a class cast error
+        setSafely((T) newValue);
+    }
+
+    void reset();
+
+    void validateValue();
+
+    void validate(T input);
+
+    T getDefault();
+
+    void format(List<String> output);
+}
diff --git a/logstash-core/src/main/java/org/logstash/settings/SettingDelegator.java b/logstash-core/src/main/java/org/logstash/settings/SettingDelegator.java
new file mode 100644
index 00000000000..bc0bf9250ea
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/SettingDelegator.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import java.util.List;
+import java.util.Objects;
+
+abstract class SettingDelegator<T> implements Setting<T> {
+    private BaseSetting<T> delegate;
+
+    /**
+     * Use this constructor to wrap another setting.
+     * */
+    SettingDelegator(BaseSetting<T> delegate) {
+        Objects.requireNonNull(delegate);
+        this.delegate = delegate;
+    }
+
+    BaseSetting<T> getDelegate() {
+        return delegate;
+    }
+
+    @Override
+    public String getName() {
+        return delegate.getName();
+    }
+
+    @Override
+    public T value() {
+        return delegate.value();
+    }
+
+    @Override
+    public boolean isSet() {
+        return delegate.isSet();
+    }
+
+    @Override
+    public boolean isStrict() {
+        return delegate.isStrict();
+    }
+
+    @Override
+    public void setSafely(T newValue) {
+        delegate.setSafely(newValue);
+    }
+
+    @Override
+    public void reset() {
+        delegate.reset();
+    }
+
+    @Override
+    public void validateValue() {
+        delegate.validateValue();
+    }
+
+    @Override
+    public T getDefault() {
+        return delegate.getDefault();
+    }
+
+    @Override
+    public void format(List<String> output) {
+        delegate.format(output);
+    }
+
+    @Override
+    public void validate(T input) {
+        delegate.validate(input);
+    }
+}
diff --git a/logstash-core/src/main/java/org/logstash/settings/SettingWithDeprecatedAlias.java b/logstash-core/src/main/java/org/logstash/settings/SettingWithDeprecatedAlias.java
new file mode 100644
index 00000000000..3b3b0678ed7
--- /dev/null
+++ b/logstash-core/src/main/java/org/logstash/settings/SettingWithDeprecatedAlias.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.logstash.settings;
+
+import java.util.Arrays;
+import java.util.List;
+
+
+/**
+ * A <code>SettingWithDeprecatedAlias</code> wraps any <code>Setting</code> to provide a deprecated
+ * alias, and hooks @see org.logstash.settings.Setting#validate_value() to ensure that a deprecation
+ * warning is fired when the setting is provided by its deprecated alias,
+ * or to produce an error when both the canonical name and deprecated
+ * alias are used together.
+ * */
+// This class is public else the getDeprecatedAlias method can't be seen from setting_with_deprecated_alias_spec.rb
+public class SettingWithDeprecatedAlias<T> extends SettingDelegator<T> {
+    
+    /**
+     * Wraps the provided setting, returning a pair of connected settings
+     * including the canonical setting and a deprecated alias.
+     * @param canonicalSetting the setting to wrap
+     * @param deprecatedAliasName the name for the deprecated alias
+     *
+     * @return List of [SettingWithDeprecatedAlias, DeprecatedAlias]
+     * */
+    static <T> List<Setting<T>> wrap(BaseSetting<T> canonicalSetting, String deprecatedAliasName) {
+        final SettingWithDeprecatedAlias<T> settingProxy = new SettingWithDeprecatedAlias<>(canonicalSetting, deprecatedAliasName);
+        return Arrays.asList(settingProxy, settingProxy.deprecatedAlias);
+    }
+
+    private DeprecatedAlias<T> deprecatedAlias;
+
+    @SuppressWarnings("this-escape")
+    protected SettingWithDeprecatedAlias(BaseSetting<T> canonicalSetting, String deprecatedAliasName) {
+        super(canonicalSetting);
+
+        this.deprecatedAlias = new DeprecatedAlias<T>(this, deprecatedAliasName);
+    }
+
+    BaseSetting<T> getCanonicalSetting() {
+        return getDelegate();
+    }
+
+    public DeprecatedAlias<T> getDeprecatedAlias() {
+        return deprecatedAlias;
+    }
+
+    @Override
+    public void setSafely(T value) {
+        getCanonicalSetting().setSafely(value);
+    }
+
+    @Override
+    public T value() {
+        if (getCanonicalSetting().isSet()) {
+            return super.value();
+        }
+        // bypass warning by querying the wrapped setting's value
+        if (deprecatedAlias.isSet()) {
+            return deprecatedAlias.getDelegate().value();
+        }
+        return getDefault();
+    }
+
+    @Override
+    public boolean isSet() {
+        return getCanonicalSetting().isSet() || deprecatedAlias.isSet();
+    }
+
+    @Override
+    public void format(List<String> output) {
+        if (!(deprecatedAlias.isSet() && !getCanonicalSetting().isSet())) {
+            super.format(output);
+            return;
+        }
+        output.add(String.format("*%s: %s (via deprecated `%s`; default: %s)",
+                getName(), value(), deprecatedAlias.getName(), getDefault()));
+    }
+
+    @Override
+    public void validateValue() {
+        if (deprecatedAlias.isSet() && getCanonicalSetting().isSet()) {
+            throw new IllegalStateException(String.format("Both `%s` and its deprecated alias `%s` have been set.\n" +
+                    "Please only set `%s`", getCanonicalSetting().getName(), deprecatedAlias.getName(), getCanonicalSetting().getName()));
+        }
+        super.validateValue();
+    }
+
+}
diff --git a/logstash-core/src/test/java/org/logstash/settings/BooleanTest.java b/logstash-core/src/test/java/org/logstash/settings/BooleanTest.java
new file mode 100644
index 00000000000..0886f6710b8
--- /dev/null
+++ b/logstash-core/src/test/java/org/logstash/settings/BooleanTest.java
@@ -0,0 +1,46 @@
+package org.logstash.settings;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class BooleanTest {
+
+
+    private Boolean sut;
+
+    @Before
+    public void setUp() {
+        sut = new Boolean("api.enabled", true);
+    }
+
+    @Test
+    public void givenLiteralBooleanStringValueWhenCoercedToBooleanValueThenIsValidBooleanSetting() {
+        sut.set("false");
+
+        Assert.assertFalse(sut.value());
+    }
+
+    @Test
+    public void givenBooleanInstanceWhenCoercedThenReturnValidBooleanSetting() {
+        sut.set(java.lang.Boolean.FALSE);
+
+        Assert.assertFalse(sut.value());
+    }
+
+    @Test
+    public void givenInvalidStringLiteralForBooleanValueWhenCoercedThenThrowsAnError() {
+        IllegalArgumentException exception = Assert.assertThrows(IllegalArgumentException.class, () -> sut.set("bananas"));
+        assertThat(exception.getMessage(), equalTo("Cannot coerce `bananas` to boolean (api.enabled)"));
+    }
+
+    @Test
+    public void givenInvalidTypeInstanceForBooleanValueWhenCoercedThenThrowsAnError() {
+        IllegalArgumentException exception = Assert.assertThrows(IllegalArgumentException.class, () -> sut.set(1));
+        assertThat(exception.getMessage(), equalTo("Cannot coerce `1` to boolean (api.enabled)"));
+    }
+
+}
\ No newline at end of file
diff --git a/qa/integration/specs/monitoring_api_spec.rb b/qa/integration/specs/monitoring_api_spec.rb
index 830ddc0bc48..19dabc16f0d 100644
--- a/qa/integration/specs/monitoring_api_spec.rb
+++ b/qa/integration/specs/monitoring_api_spec.rb
@@ -389,14 +389,14 @@
     end
 
     #default
-    logging_get_assert logstash_service, "INFO", "TRACE",
+    logging_get_assert logstash_service, ["WARN", "INFO"], "TRACE",
                        skip: 'logstash.licensechecker.licensereader' #custom (ERROR) level to start with
 
     #root logger - does not apply to logger.slowlog
     logging_put_assert logstash_service.monitoring_api.logging_put({"logger." => "WARN"})
     logging_get_assert logstash_service, "WARN", "TRACE"
     logging_put_assert logstash_service.monitoring_api.logging_put({"logger." => "INFO"})
-    logging_get_assert logstash_service, "INFO", "TRACE"
+    logging_get_assert logstash_service, ["WARN", "INFO"], "TRACE"
 
     #package logger
     logging_put_assert logstash_service.monitoring_api.logging_put({"logger.logstash.agent" => "DEBUG"})
@@ -422,7 +422,7 @@
 
     # all log levels should be reset to original values
     logging_put_assert logstash_service.monitoring_api.logging_reset
-    logging_get_assert logstash_service, "INFO", "TRACE"
+    logging_get_assert logstash_service, ["WARN", "INFO"], "TRACE"
   end
 
 
@@ -433,7 +433,15 @@ def logging_get_assert(logstash_service, logstash_level, slowlog_level, skip: ''
     result["loggers"].each do |k, v|
       next if !k.empty? && k.eql?(skip)
       if k.start_with? "logstash", "org.logstash" #logstash is the ruby namespace, and org.logstash for java
-        expect(v).to eq(logstash_level), "logstash logger '#{k}' has logging level: #{v} expected: #{logstash_level}"
+        if logstash_level.is_a?(Array)
+          if logstash_level.size == 1
+            expect(v).to eq(logstash_level[0]), "logstash logger '#{k}' has logging level: #{v} expected: #{logstash_level[0]}"
+          else
+            expect(logstash_level).to include(v), "logstash logger '#{k}' has logging level: #{v} expected to be one of: #{logstash_level}"
+          end
+        else
+          expect(v).to eq(logstash_level), "logstash logger '#{k}' has logging level: #{v} expected: #{logstash_level}"
+        end
       elsif k.start_with? "slowlog"
         expect(v).to eq(slowlog_level), "slowlog logger '#{k}' has logging level: #{v} expected: #{slowlog_level}"
       end